《Spark Python API 官方文档中文版》 之 pyspark.sql (二)

摘要:在Spark开发中,由于需要用Python实现,发现API与Scala的略有不同,而Python API的中文资料相对很少。每次去查英文版API的说明相对比较慢,还是中文版比较容易get到所需,所以利用闲暇之余将官方文档翻译为中文版,并亲测Demo的代码。在此记录一下,希望对那些对Spark感兴趣和从事大数据开发的人员提供有价值的中文资料,对PySpark开发人员的工作和学习有所帮助。

官网地址:http://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html            

pyspark.sql module

Module Context

Spark SQL和DataFrames重要的类有:
pyspark.sql.SQLContext DataFrame和SQL方法的主入口
pyspark.sql.DataFrame 将分布式数据集分组到指定列名的数据框中
pyspark.sql.Column DataFrame中的列
pyspark.sql.Row DataFrame数据的行
pyspark.sql.HiveContext 访问Hive数据的主入口
pyspark.sql.GroupedData 由DataFrame.groupBy()创建的聚合方法集
pyspark.sql.DataFrameNaFunctions 处理丢失数据(空数据)的方法
pyspark.sql.DataFrameStatFunctions 统计功能的方法
pyspark.sql.functions DataFrame可用的内置函数
pyspark.sql.types 可用的数据类型列表
pyspark.sql.Window 用于处理窗口函数

3.class pyspark.sql.DataFrame(jdf, sql_ctx)

分布式的收集数据分组到命名列中。
一个DataFrame相当于在Spark SQL中一个相关的表,可在SQLContext使用各种方法创建,如:

people = sqlContext.read.parquet("...")

一旦创建, 可以使用在DataFrame、Column中定义的不同的DSL方法操作。
从data frame中返回一列使用对应的方法:

ageCol = people.age

一个更具体的例子:

# To create DataFrame using SQLContext
people = sqlContext.read.parquet("...")
department = sqlContext.read.parquet("...") people.filter(people.age > 30).join(department, people.deptId == department.id)).groupBy(department.name, "gender").agg({"salary": "avg", "age": "max"})

3.1 agg(*exprs)

没有组的情况下聚集整个DataFrame (df.groupBy.agg()的简写)。

>>> l=[('jack',5),('john',4),('tom',2)]
>>> df = sqlContext.createDataFrame(l,['name','age']) >>> df.agg({"age": "max"}).collect() [Row(max(age)=5)] >>> from pyspark.sql import functions as F >>> df.agg(F.min(df.age)).collect() [Row(min(age)=2)]

3.2 alias(alias)

返回一个设置别名的新的DataFrame。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age']) >>> from pyspark.sql.functions import * >>> df_as1 = df.alias("df_as1") >>> df_as2 = df.alias("df_as2") >>> joined_df = df_as1.join(df_as2, col("df_as1.name") == col("df_as2.name"), 'inner') >>> joined_df.select(col("df_as1.name"), col("df_as2.name"), col("df_as2.age")).collect() [Row(name=u'Alice', name=u'Alice', age=2), Row(name=u'Bob', name=u'Bob', age=5)]

3.3 cache()

用默认的存储级别缓存数据(MEMORY_ONLY_SER).

3.4 coalesce(numPartitions)

返回一个有确切的分区数的分区的新的DataFrame。
与在一个RDD上定义的合并类似, 这个操作产生一个窄依赖。 如果从1000个分区到100个分区,不会有shuffle过程, 而是每100个新分区会需要当前分区的10个。

>>> df.coalesce(1).rdd.getNumPartitions()
1

3.5 collect()

返回所有的记录数为行的列表。

>>> df.collect()
[Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]

3.6 columns

返回所有列名的列表。

>>> df.columns
['age', 'name']

3.7 corr(col1, col2, method=None)

计算一个DataFrame相关的两列为double值。通常只支持皮尔森相关系数。DataFrame.corr()和DataFrameStatFunctions.corr()类似。
参数:●  col1 – 第一列的名称  
      ●  col2 – 第二列的名称
           ●  method – 相关方法.当前只支持皮尔森相关系数

3.8 count()

返回DataFrame的行数。

>>> df.count()
2

3.9 cov(col1, col2)

计算由列名指定列的样本协方差为double值。DataFrame.cov()和DataFrameStatFunctions.cov()类似。
参数:●  col1 – 第一列的名称
      ●  col2 – 第二列的名称

3.10 crosstab(col1, col2)

计算给定列的分组频数表,也称为相关表。每一列的去重值的个数应该小于1e4.最多返回1e6个非零对.每一行的第一列会是col1的去重值,列名称是col2的去重值。第一列的名称是$col1_$col2. 没有出现的配对将以零作为计数。DataFrame.crosstab() and DataFrameStatFunctions.crosstab()类似。
参数:●  col1 – 第一列的名称. 去重项作为每行的第一项。
      ●  col2 – 第二列的名称. 去重项作为DataFrame的列名称。

3.11 cube(*cols)

创建使用指定列的当前DataFrame的多维立方体,这样可以聚合这些数据。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.cube('name', df.age).count().show()
+-----+----+-----+
| name| age|count|
+-----+----+-----+
| null|   2|    1|
|Alice|null|    1|
|  Bob|   5|    1|
|  Bob|null|    1|
| null|   5|    1|
| null|null|    2|
|Alice|   2|    1|
+-----+----+-----+

3.12 describe(*cols)

计算数值列的统计信息。
包括计数,平均,标准差,最小和最大。如果没有指定任何列,这个函数计算统计所有数值列。

>>> df.describe().show()
+-------+------------------+
|summary|               age|
+-------+------------------+
|  count|                 2|
|   mean|               3.5|
| stddev|2.1213203435596424|
|    min|                 2|
|    max|                 5|
+-------+------------------+
>>> df.describe(['age', 'name']).show()
+-------+------------------+-----+
|summary|               age| name|
+-------+------------------+-----+
|  count|                 2|    2|
|   mean|               3.5| null|
| stddev|2.1213203435596424| null|
|    min|                 2|Alice|
|    max|                 5|  Bob|
+-------+------------------+-----+

3.13 distinct()

返回行去重的新的DataFrame。

>>> l=[('Alice',2),('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.distinct().count()
2

3.14 drop(col)

返回删除指定列的新的DataFrame。
参数:●  col – 要删除列的字符串类型名称,或者要删除的列。

>>> df.drop('age').collect()
[Row(name=u'Alice'), Row(name=u'Bob')] 
>>> df.drop(df.age).collect()
[Row(name=u'Alice'), Row(name=u'Bob')]
>>> l1=[('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> l2=[('Bob',85)]
>>> df2 = sqlContext.createDataFrame(l2,['name','height'])
>>> df.join(df2, df.name == df2.name, 'inner').drop(df.name).collect()
[Row(age=5, height=85, name=u'Bob')]
>>> df.join(df2, df.name == df2.name, 'inner').drop(df2.name).collect()
[Row(age=5, name=u'Bob', height=85)]

3.15 dropDuplicates(subset=None)

返回去掉重复行的一个新的DataFrame,通常只考虑某几列。
drop_duplicates()和dropDuplicates()类似。

>>> from pyspark.sql import Row
>>> df = sc.parallelize([Row(name='Alice', age=5, height=80),Row(name='Alice', age=5, height=80),Row(name='Alice', age=10, height=80)]).toDF()
>>> df.dropDuplicates().show()
+---+------+-----+
|age|height| name|
+---+------+-----+
|  5|    80|Alice|
| 10|    80|Alice|
+---+------+-----+
>>> df.dropDuplicates(['name', 'height']).show()
+---+------+-----+
|age|height| name|
+---+------+-----+
|  5|    80|Alice|
+---+------+-----+

3.16 drop_duplicates(subset=None)

与以上相同。

3.17 dropna(how=’any’, thresh=None, subset=None)

返回一个删除null值行的新的DataFrame。dropna()和dataframenafunctions.drop()类似。
参数:●  how – ‘any’或者’all’。如果’any’,删除包含任何空值的行。如果’all’,删除所有值为null的行。
    thresh – int,默认为None,如果指定这个值,删除小于阈值的非空值的行。这个会重写’how’参数。
   ●  subset – 选择的列名称列表。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> dfnew = df.cube('name', df.age).count()
>>> dfnew.show()
+-----+----+-----+
| name| age|count|
+-----+----+-----+
| null|   2|    1|
|Alice|null|    1|
|  Bob|   5|    1|
|  Bob|null|    1|
| null|   5|    1|
| null|null|    2|
|Alice|   2|    1|
+-----+----+-----+
>>> dfnew.na.drop().show()
+-----+---+-----+
| name|age|count|
+-----+---+-----+
|  Bob|  5|    1|
|Alice|  2|    1|
+-----+---+-----+

3.18 dtypes

返回所有列名及类型的列表。

>>> df.dtypes
[('age', 'int'), ('name', 'string')]

3.19 explain(extended=False)

将(逻辑和物理)计划打印到控制台以进行调试。
参数:●  extended – boolean类型,默认为False。如果为False,只打印物理计划。

>>> df.explain()
== Physical Plan ==
Scan ExistingRDD[age#0,name#1]
>>> df.explain(True)
== Parsed Logical Plan ==
...
== Analyzed Logical Plan ==
...
== Optimized Logical Plan ==
...
== Physical Plan ==
...

3.20 fillna(value, subset=None)

替换空值,和na.fill()类似,DataFrame.fillna()和dataframenafunctions.fill()类似。
参数:●  value – 要代替空值的值有int,long,float,string或dict.如果值是字典,subset参数将被忽略。值必须是要替换的列的映射,替换值必须是int,long,float或者string.
      ●  subset – 要替换的列名列表。在subset指定的列,没有对应数据类型的会被忽略。例如,如果值是字符串,subset包含一个非字符串的列,这个非字符串的值会被忽略。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> dfnew = df.cube('name', df.age).count()
>>> dfnew.show()
+-----+----+-----+
| name| age|count|
+-----+----+-----+
| null|   2|    1|
|Alice|null|    1|
|  Bob|   5|    1|
|  Bob|null|    1|
| null|   5|    1|
| null|null|    2|
|Alice|   2|    1|
+-----+----+-----+
>>> dfnew.na.fill(50).show()
+-----+---+-----+
| name|age|count|
+-----+---+-----+
| null|  2|    1|
|Alice| 50|    1|
|  Bob|  5|    1|
|  Bob| 50|    1|
| null|  5|    1|
| null| 50|    2|
|Alice|  2|    1|
+-----+---+-----+
>>> dfnew.na.fill({'age': 50, 'name': 'unknown'}).show()
+-------+---+-----+
|   name|age|count|
+-------+---+-----+
|unknown|  2|    1|
|  Alice| 50|    1|
|    Bob|  5|    1|
|    Bob| 50|    1|
|unknown|  5|    1|
|unknown| 50|    2|
|  Alice|  2|    1|
+-------+---+-----+

3.21 filter(condition)

用给定的条件过滤行。
where()和filter()类似。
参数:●  条件 – 一个列的bool类型或字符串的SQL表达式。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.filter(df.age > 3).collect()
[Row(age=5, name=u'Bob')]
>>> df.where(df.age == 2).collect()
[Row(age=2, name=u'Alice')]
>>> df.filter("age > 3").collect()
[Row(age=5, name=u'Bob')]
>>> df.where("age = 2").collect()
[Row(age=2, name=u'Alice')]

3.22 first()

返回第一行。

>>> df.first()
Row(age=2, name=u'Alice')

3.23 flatMap(f)

返回在每行应用F函数后的新的RDD,然后将结果压扁。
是df.rdd.flatMap()的简写。

>>> df.flatMap(lambda p: p.name).collect()
[u'A', u'l', u'i', u'c', u'e', u'B', u'o', u'b']

3.24 foreach(f)

应用f函数到DataFrame的所有行。
是df.rdd.foreach()的简写。

>>> def f(person):
...     print(person.name)
>>> df.foreach(f)
Alice
Bob

3.25 foreachPartition(f)

应用f函数到DataFrame的每一个分区。
是 df.rdd.foreachPartition()的缩写。

>>> def f(people):
...     for person in people:
...         print(person.name)
>>> df.foreachPartition(f)
Alice
Bob

3.26 freqItems(cols, support=None)

参数:●  cols – 要计算重复项的列名,为字符串类型的列表或者元祖。
      ●  support – 要计算频率项的频率值。默认是1%。参数必须大于1e-4.

3.27 groupBy(*cols)

使用指定的列分组DataFrame,这样可以聚合计算。可以从GroupedData查看所有可用的聚合方法。
groupby()和groupBy()类似。
参数:●  cols – 分组依据的列。每一项应该是一个字符串的列名或者列的表达式。

>>> df.groupBy().avg().collect()
[Row(avg(age)=3.5)]
>>> df.groupBy('name').agg({'age': 'mean'}).collect()
[Row(name=u'Alice', avg(age)=2.0), Row(name=u'Bob', avg(age)=5.0)]
>>> df.groupBy(df.name).avg().collect()
[Row(name=u'Alice', avg(age)=2.0), Row(name=u'Bob', avg(age)=5.0)]
>>> df.groupBy(['name', df.age]).count().collect()
[Row(name=u'Bob', age=5, count=1), Row(name=u'Alice', age=2, count=1)]

3.28 groupby(*cols)

和以上一致

3.29 head(n=None)

返回前n行
参数:●  n – int类型,默认为1,要返回的行数。
返回值: 如果n大于1,返回行列表,如果n为1,返回单独的一行。

>>> df.head()
Row(age=2, name=u'Alice')
>>> df.head(1)
[Row(age=2, name=u'Alice')]

3.30 insertInto(tableName, overwrite=False)

插入DataFrame内容到指定表。
注:在1.4中已过时,使用DataFrameWriter.insertInto()代替。

3.31 intersect(other)

返回新的DataFrame,包含仅同时在当前框和另一个框的行。
相当于SQL中的交集。

3.32 intersect(other)

如果collect()和take()方法可以运行在本地(不需要Spark executors)那么返回True

3.33 join(other, on=None, how=None)

使用给定的关联表达式,关联另一个DataFrame。
以下执行df1和df2之间完整的外连接。
参数:● other – 连接的右侧
   ● on – 一个连接的列名称字符串, 列名称列表,一个连接表达式(列)或者列的列表。如果on参数是一个字符串或者字符串列表,表示连接列的名称,这些名称必须同时存在join的两个表中, 这样执行的是一个等价连接。
   ● how – 字符串,默认’inner’。inner,outer,left_outer,right_outer,leftsemi之一。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> l2=[('Tom',80),('Bob',85)]
>>> df2 = sqlContext.createDataFrame(l2,['name','height'])
>>> df.join(df2, df.name == df2.name, 'outer').select(df.name, df2.height).collect()
[Row(name=None, height=80), Row(name=u'Alice', height=None), Row(name=u'Bob', height=85)]
>>> df.join(df2, 'name', 'outer').select('name', 'height').collect()
[Row(name=u'Tom', height=80), Row(name=u'Alice', height=None), Row(name=u'Bob', height=85)]
>>> l3=[('Alice',2,60),('Bob',5,80)]
>>> df3 = sqlContext.createDataFrame(l3,['name','age','height'])
>>> cond = [df.name == df3.name, df.age == df3.age]
>>> df.join(df3, cond, 'outer').select(df.name, df3.age).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.join(df2, 'name').select(df.name, df2.height).collect()
[Row(name=u'Bob', height=85)]
>>> l4=[('Alice',1),('Bob',5)]
>>> df4 = sqlContext.createDataFrame(l4,['name','age'])
>>> df.join(df4, ['name', 'age']).select(df.name, df.age).collect()
[Row(name=u'Bob', age=5)]

3.34 limit(num)

将结果计数限制为指定的数字。

>>> df.limit(1).collect()
[Row(age=2, name=u'Alice')]
>>> df.limit(0).collect()
[]

3.35 map(f)

通过每行应用f函数返回新的RDD。
是 df.rdd.map()的缩写。

>>> df.map(lambda p: p.name).collect()
[u'Alice', u'Bob']

3.36 mapPartitions(f, preservesPartitioning=False)

通过每个分区应用f函数返回新的RDD
是df.rdd.mapPartitions()的缩写。

>>> rdd = sc.parallelize([1, 2, 3, 4], 4)
>>> def f(iterator): yield 1
...
>>> rdd.mapPartitions(f).sum()
4

3.37 na

返回DataFrameNaFunctions用于处理缺失值。

3.38 orderBy(*cols, **kwargs)

返回按照指定列排序的新的DataFrame。
参数:● cols – 用来排序的列或列名称的列表。
      ● ascending – 布尔值或布尔值列表(默认 True). 升序排序与降序排序。指定多个排序顺序的列表。如果指定列表, 列表的长度必须等于列的长度。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.sort(df.age.desc()).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.sort("age", ascending=False).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.orderBy(df.age.desc()).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> from pyspark.sql.functions import *
>>> df.sort(asc("age")).collect()
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
>>> df.orderBy(desc("age"), "name").collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.orderBy(["age", "name"], ascending=[0, 1]).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]

3.39 persist(storageLevel=StorageLevel(False, True, False, False, 1))

设置存储级别以在第一次操作运行完成后保存其值。这只能用来分配新的存储级别,如果RDD没有设置存储级别的话。如果没有指定存储级别,默认为(memory_only_ser)。

3.40 printSchema()

打印schema以树的格式

>>> df.printSchema()
root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)

3.41 randomSplit(weights, seed=None)

按照提供的权重随机的划分DataFrame。
参数:● weights – doubles类型的列表做为权重来划分DataFrame。权重会被恢复如果总值不到1.0。
       seed – random的随机数。

>>> l4=[('Alice',1),('Bob',5),('Jack',8),('Tom',10)]
>>> df4 = sqlContext.createDataFrame(l4,['name','age'])
>>> splits = df4.randomSplit([1.0, 2.0],24)
>>> splits[0].count()
1
>>> splits[1].count()
3

3.42 rdd

返回内容为行的RDD。

3.43 registerAsTable(name)

注:在1.4中已过时,使用registerTempTable()代替。

3.44 registerTempTable(name)

使用给定的名字注册该RDD为临时表
这个临时表的有效期与用来创建这个DataFrame的SQLContext相关

>>> df.registerTempTable("people")
>>> df2 = sqlContext.sql("select * from people")
>>> sorted(df.collect()) == sorted(df2.collect())
True

3.45 repartition(numPartitions, *cols)

按照给定的分区表达式分区,返回新的DataFrame。产生的DataFrame是哈希分区。
numPartitions参数可以是一个整数来指定分区数,或者是一个列。如果是一个列,这个列会作为第一个分区列。如果没有指定,将使用默认的分区数。
1.6版本修改: 添加可选参数可以指定分区列。如果分区列指定的话,numPartitions也是可选的。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.repartition(10).rdd.getNumPartitions()
10
>>> data = df.unionAll(df).repartition("age")
>>> data.show()
+-----+---+
| name|age|
+-----+---+
|Alice|  2|
|Alice|  2|
|  Bob|  5|
|  Bob|  5|
+-----+---+
>>> data = data.repartition(7, "age")
>>> data.show()
+-----+---+
| name|age|
+-----+---+
|  Bob|  5|
|  Bob|  5|
|Alice|  2|
|Alice|  2|
+-----+---+
>>> data.rdd.getNumPartitions()
7
>>> data = data.repartition("name", "age")
>>> data.show()
+-----+---+
| name|age|
+-----+---+
|  Bob|  5|
|  Bob|  5|
|Alice|  2|
|Alice|  2|
+-----+---+

3.46 replace(to_replace, value, subset=None)

返回用另外一个值替换了一个值的新的DataFrame。DataFrame.replace() 和 DataFrameNaFunctions.replace() 类似。
参数:● to_replace – 整形,长整形,浮点型,字符串,或者列表。要替换的值。如果值是字典,那么值会被忽略,to_replace必须是一个从列名(字符串)到要替换的值的映射。要替换的值必须是一个整形,长整形,浮点型,或者字符串。
           ● value – 整形,长整形,浮点型,字符串或者列表。要替换为的值。要替换为的值必须是一个整形,长整形,浮点型,或者字符串。如果值是列表或者元组,值应该和to_replace有相同的长度。
           ● subset – 要考虑替换的列名的可选列表。在subset指定的列如果没有匹配的数据类型那么将被忽略。例如,如果值是字符串,并且subset参数包含一个非字符串的列,那么非字符串的列被忽略。

>>> l4=[('Alice',10,80),('Bob',5,None),('Tom',None,None),(None,None,None)]
>>> df4 = sqlContext.createDataFrame(l4,['name','age','height'])
>>> df4.na.replace(10, 20).show()
+-----+----+------+
| name| age|height|
+-----+----+------+
|Alice|  20|    80|
|  Bob|   5|  null|
|  Tom|null|  null|
| null|null|  null|
+-----+----+------+
>>> df4.na.replace(['Alice', 'Bob'], ['A', 'B'], 'name').show()
+----+----+------+
|name| age|height|
+----+----+------+
|   A|  10|    80|
|   B|   5|  null|
| Tom|null|  null|
|null|null|  null|
+----+----+------+

3.47 rollup(*cols)

使用指定的列为当前的DataFrame创建一个多维汇总, 这样可以聚合这些数据。

>>> l=[('Alice',2,80),('Bob',5,None)]
>>> df = sqlContext.createDataFrame(l,['name','age','height'])
>>> df.rollup('name', df.age).count().show()
+-----+----+-----+
| name| age|count|
+-----+----+-----+
|Alice|null|    1|
|  Bob|   5|    1|
|  Bob|null|    1|
| null|null|    2|
|Alice|   2|    1|
+-----+----+-----+

3.48 sample(withReplacement, fraction, seed=None)

返回DataFrame的子集采样。

>>> df.sample(False, 0.5, 42).count()
2

3.49 sampleBy(col, fractions, seed=None)

根据每个层次上给出的分数,返回没有替换的分层样本。
返回没有替换的分层抽样 基于每层给定的一小部分 在给定的每层的片段
参数:● col – 定义层的列
           ● fractions – 每层的抽样数。如果没有指定层, 将其数目视为0.
           ● seed – 随机数
返回值: 返回代表分层样本的新的DataFrame

>>> from pyspark.sql.functions import col
>>> dataset = sqlContext.range(0, 100).select((col("id") % 3).alias("key"))
>>> sampled = dataset.sampleBy("key", fractions={0: 0.1, 1: 0.2}, seed=0)
>>> sampled.groupBy("key").count().orderBy("key").show()
+---+-----+
|key|count|
+---+-----+
|  0|    5|
|  1|    9|
+---+-----+

3.50 save(path=None, source=None, mode=’error’, **options)

保存DataFrame的数据到数据源。
注:在1.4中已过时,使用DataFrameWriter.save()代替。

3.51 saveAsParquetFile(path)

保存内容为一个Parquet文件,代表这个schema。
注:在1.4中已过时,使用DataFrameWriter.parquet() 代替。

3.52 saveAsTable(tableName, source=None, mode=’error’, **options)

将此DataFrame的内容作为表保存到数据源。
注:在1.4中已过时,使用DataFrameWriter.saveAsTable() 代替。

3.53  schema

返回DataFrame的schema为types.StructType。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.schema
StructType(List(StructField(name,StringType,true),StructField(age,LongType,true)))

3.54  select(*cols)

提供一组表达式并返回一个新的DataFrame。
参数:● cols – 列名(字符串)或表达式(列)列表。 如果其中一列的名称为“*”,那么该列将被扩展为包括当前DataFrame中的所有列。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.select('*').collect()
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
>>> df.select('name', 'age').collect()
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
>>> df.select(df.name, (df.age + 10).alias('age')).collect()
[Row(name=u'Alice', age=12), Row(name=u'Bob', age=15)]

3.55 selectExpr(*expr)

投射一组SQL表达式并返回一个新的DataFrame。
这是接受SQL表达式的select()的变体。

>>> df.selectExpr("age * 2", "abs(age)").collect()
[Row((age * 2)=4, abs(age)=2), Row((age * 2)=10, abs(age)=5)]

3.56 show(n=20, truncate=True) 

将前n行打印到控制台。
参数:● n – 要显示的行数。
           ● truncate – 是否截断长字符串并对齐单元格。

>>> df
DataFrame[name: string, age: bigint]
>>> df.show()
+-----+---+
| name|age|
+-----+---+
|Alice|  2|
|  Bob|  5|
+-----+---+

3.57 sort(*cols, **kwargs)

返回按指定列排序的新DataFrame。
参数:● cols – 要排序的列或列名称列表。
           ● ascending – 布尔值或布尔值列表(默认为True)。 排序升序降序。 指定多个排序顺序的列表。 如果指定了列表,列表的长度必须等于列的长度。

>>> df.sort(df.age.desc()).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.sort("age", ascending=False).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.orderBy(df.age.desc()).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> from pyspark.sql.functions import *
>>> df.sort(asc("age")).collect()
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
>>> df.orderBy(desc("age"), "name").collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]
>>> df.orderBy(["age", "name"], ascending=[0, 1]).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]

3.58 sortWithinPartitions(*cols, **kwargs)

返回一个新的DataFrame,每个分区按照指定的列排序。
参数:● cols – 要排序的列或列名称列表。
            ascending – 布尔值或布尔值列表(默认为True)。 排序升序降序。 指定多个排序顺序的列表。 如果指定了列表,列表的长度必须等于列的长度。

>>> df.sortWithinPartitions("age", ascending=False).show()
+-----+---+
| name|age|
+-----+---+
|Alice|  2|
|  Bob|  5|
+-----+---+

3.59 stat 

返回统计功能的DataFrameStatFunctions。

3.60 subtract(other)

返回一个新的DataFrame,这个DataFrame中包含的行不在另一个DataFrame中。
这相当于SQL中的EXCEPT。

3.61 take(num)

返回前num行的行列表

>>> df.take(2)
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]

3.62 toDF(*cols)

返回一个新类:具有新的指定列名称的DataFrame。
参数:● cols – 新列名列表(字符串)。

>>> df.toDF('f1', 'f2').collect()
[Row(f1=u'Alice', f2=2), Row(f1=u'Bob', f2=5)]

3.63 toJSON(use_unicode=True)

将DataFrame转换为字符串的RDD。
每行都将转换为JSON格式作为返回的RDD中的一个元素。

>>> df.toJSON().first()
u'{"name":"Alice","age":2}'

3.64 toPandas()

将此DataFrame的内容返回为Pandas pandas.DataFrame。
这只有在pandas安装和可用的情况下才可用。

>>> df.toPandas()  
   age   name
0    2  Alice
1    5    Bob

3.65 unionAll(other)

返回包含在这个frame和另一个frame的行的联合的新DataFrame。
这相当于SQL中的UNION ALL。

3.66 unpersist(blocking=True)

将DataFrame标记为非持久性,并从内存和磁盘中删除所有的块。

3.67 where(condition)

使用给定表达式过滤行。
where()是filter()的别名。
参数:● condition – 一个布尔类型的列或一个SQL表达式的字符串。

>>> l=[('Alice',2),('Bob',5)]
>>> df = sqlContext.createDataFrame(l,['name','age'])
>>> df.filter(df.age > 3).collect()
[Row(name=u'Bob', age=5)]
>>> df.where(df.age == 2).collect()
[Row(name=u'Alice', age=2)]

>>> df.filter("age > 3").collect()
[Row(name=u'Bob', age=5)]
>>> df.where("age = 2").collect()
[Row(name=u'Alice', age=2)]

3.68 withColumn(colName, col)

通过添加列或替换具有相同名称的现有列来返回新的DataFrame。
参数:● colName – 字符串,新列的名称
           ● col – 新列的列表达式

>>> df.withColumn('age2', df.age + 2).collect()
[Row(name=u'Alice', age=2, age2=4), Row(name=u'Bob', age=5, age2=7)]

3.69 withColumnRenamed(existing, new)

通过重命名现有列来返回新的DataFrame。
参数:● existing – 字符串,要重命名的现有列的名称
           ● col – 字符串,列的新名称

>>> df.withColumnRenamed('age', 'age2').collect()
[Row(name=u'Alice', age2=2), Row(name=u'Bob', age2=5)]

3.70 write

用于将DataFrame的内容保存到外部存储的接口。
返回:DataFrameWriter

 

    原文作者:spark
    原文地址: https://www.cnblogs.com/wonglu/p/7784825.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞