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

Posted 2020-10-13

摘要：在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 用于处理窗口函数

1.class pyspark.sql.SQLContext(sparkContext, sqlContext=None)

SQLContext可以用来创建DataFrame、注册DataFrame为表、在表上执行SQL、缓存表、读取parquet文件。

参数：●  sparkContext - 支持sqlcontext的sparkcontext
           ●  sqlContext - 一个可选的JVM Scala sqlcontext，若设置，我们不需要在JVM实例化一个新的sqlcontext，而是都调用这个对象。

1.1 applySchema(rdd, schema)

注：在1.3中已过时,使用createDataFrame()代替。

1.2 cacheTable(tableName)

缓存表到内存中

1.3 clearCache()

从内存缓存删除所有缓存表。

1.4 createDataFrame(data, schema=None, samplingRatio=None)

从元组/列表RDD或列表或pandas.DataFrame创建DataFrame
当模式是列名的列表时，每个列的类型会从数据中推断出来。
当模式没有时，将尝试从数据中推断模式（列名和类型）,数据应该是行或命名元组或字典的RDD。
如果模式推理是必要的，samplingRatio用来确定用于模式推理的行比率。如果没有samplingratio，将使用第一行。

参数：●  data - 行或元组或列表或字典的RDD、list、pandas.DataFrame.
　　    ● schema – 一个结构化类型或者列名列表，默认是空。

samplingRatio – 用于推断的行的样本比率。
返回： DataFrame

>>> l=[(‘Alice‘,1)]
>>> sqlContext.createDataFrame(l).collect()
[Row(_1=u‘Alice‘, _2=1)]
>>> sqlContext.createDataFrame(l,[‘name‘,‘age‘]).collect()
[Row(name=u‘Alice‘, age=1)]

>>> d=[{‘name‘:‘Alice‘,‘age‘:1}]
>>> sqlContext.createDataFrame(d).collect()
[Row(age=1, name=u‘Alice‘)]

>>> rdd=sc.parallelize(l)
>>> sqlContext.createDataFrame(rdd).collect()
[Row(_1=u‘Alice‘, _2=1)]
>>> df=sqlContext.createDataFrame(rdd,[‘name‘,‘age‘])
>>> df.collect()
[Row(name=u‘Alice‘, age=1)]

>>> sqlContext.createDataFrame(df.toPandas()).collect()  
[Row(name=u‘Alice‘, age=1)]
>>> sqlContext.createDataFrame(pandas.DataFrame([[1, 2]])).collect()  
[Row(0=1, 1=2)]

1.5 createExternalTable(tableName, path=None, source=None, schema=None, **options)

创建基于数据源中的数据的外部表.
返回与外部表关联的DataFrame
数据源由源和一组选项指定。如果未指定源，那么将使用由spark.sql.sources.default 配置的默认的数据源配置。
通常，一个模式可以被提供作为返回的DataFrame的模式，然后创建外部表。
返回： DataFrame

1.6 dropTempTable(tableName)

从目录中删除临时表

>>> sqlContext.registerDataFrameAsTable(df, "table1")
>>> sqlContext.dropTempTable("table1")

1.7 getConf(key, defaultValue)

返回指定键的Spark SQL配置属性值。
如果键没有指定返回默认值。

1.8 inferSchema(rdd, samplingRatio=None)

注：在1.3中已过时,使用createDataFrame()代替。

1.9 jsonFile(path, schema=None, samplingRatio=1.0)

从一个文本文件中加载数据，这个文件的每一行均为JSON字符串。
注：在1.4中已过时,使用DataFrameReader.json()代替。

1.10 jsonRDD(rdd, schema=None, samplingRatio=1.0)

从一个已经存在的RDD中加载数据，这个RDD中的每一个元素均为一个JSON字符串。
如果提供了模式，将给定的模式应用到这个JSON数据集。否则，它根据数据集的采样比例来确定模式。

>>> json=sc.parallelize(["""{"name":"jack","addr":{"city":"beijing","mail":"10001"}}""","""{"name":"john","addr":{"city":"shanghai","mail":"10002"}}"""])
>>> df1 = sqlContext.jsonRDD(json)
>>> df1.collect()
[Row(addr=Row(city=u‘beijing‘, mail=u‘10001‘), name=u‘jack‘), Row(addr=Row(city=u‘shanghai‘, mail=u‘10002‘), name=u‘john‘)]

>>> df2 = sqlContext.jsonRDD(json,df1.schema)
>>> df2.collect()
[Row(addr=Row(city=u‘beijing‘, mail=u‘10001‘), name=u‘jack‘), Row(addr=Row(city=u‘shanghai‘, mail=u‘10002‘), name=u‘john‘)]

1.11 load(path=None, source=None, schema=None, **options)

返回数据源中的数据集为DataFrame.
注：在1.4中已过时,使用DataFrameReader.load()代替。

1.12 newSession()

返回一个新的SQLContext做为一个新的会话,这个会话有单独的SQLConf,注册临时表和UDFs，但共享sparkcontext和缓存表。

1.13 parquetFile(*paths)

加载Parquet文件，返回结果为DataFrame
注：在1.4中已过时,使用DataFrameReader.parquet()代替。

1.14 range(start, end=None, step=1, numPartitions=None)

创建只有一个名为id的长类型的列的DataFrame,包含从开始到结束的按照一定步长的独立元素。

参数：●  start - 开始值
　　    ●  end -  结束值
           ●  step - 增量值(默认：1)            
　　　●  numPartitions – DataFrame分区数

返回: DataFrame

>>> sqlContext.range(1, 7, 2).collect()
[Row(id=1), Row(id=3), Row(id=5)]

如果仅有一个参数，那么这个参数被作为结束值。

>>> sqlContext.range(3).collect()
[Row(id=0), Row(id=1), Row(id=2)]

1.15 read

返回一个DataFrameReader，可用于读取数据为DataFrame。

1.16 registerDataFrameAsTable(df, tableName)

注册给定的DataFrame作为目录中的临时表。
临时表只在当前SQLContext实例有效期间存在。

>>> sqlContext.registerDataFrameAsTable(df, "table1")

1.17 registerFunction(name, f, returnType=StringType)

注册python方法(包括lambda方法)，作为UDF,这样可以在 SQL statements中使用。
除了名称和函数本身之外，还可以选择性地指定返回类型。当返回类型没有指定时，默认自动转换为字符串。对于任何其他返回类型，所生成的对象必须与指定的类型匹配。
参数：●  name - UDF名称
　　    ●  f – python方法
　　    ●  返回类型 数据类型对象

>>> sqlContext.registerFunction("stringLengthString", lambda x: len(x))
>>> sqlContext.sql("SELECT stringLengthString(‘test‘)").collect()
[Row(_c0=u‘4‘)]

>>> from pyspark.sql.types import IntegerType
>>> sqlContext.registerFunction("stringLengthInt", lambda x: len(x), IntegerType())
>>> sqlContext.sql("SELECT stringLengthInt(‘test‘)").collect()
[Row(_c0=4)]

>>> from pyspark.sql.types import IntegerType
>>> sqlContext.udf.register("stringLengthInt", lambda x: len(x), IntegerType())
>>> sqlContext.sql("SELECT stringLengthInt(‘test‘)").collect()
[Row(_c0=4)]

1.18 setConf(key, value)

设置给定的Spark SQL配置属性

1.19 sql(sqlQuery)

返回DataFrame代表给定查询的结果
参数：● sqlQuery - sql语句  
返回: DataFrame

>>> l=[(1,‘row1‘),(2,‘row2‘),(3,‘row3‘)]
>>> df = sqlContext.createDataFrame(l,[‘field1‘,‘field2‘])
>>> sqlContext.registerDataFrameAsTable(df, "table1")
>>> df2 = sqlContext.sql("SELECT field1 AS f1, field2 as f2 from table1")
>>> df2.collect()
[Row(f1=1, f2=u‘row1‘), Row(f1=2, f2=u‘row2‘), Row(f1=3, f2=u‘row3‘)]

1.20 table(tableName)

返回指定的表为DataFrame
返回: DataFrame

>>> l=[(1,‘row1‘),(2,‘row2‘),(3,‘row3‘)]
>>> df = sqlContext.createDataFrame(l,[‘field1‘,‘field2‘])
>>> sqlContext.registerDataFrameAsTable(df, "table1")
>>> df2 = sqlContext.table("table1")
>>> sorted(df.collect()) == sorted(df2.collect())
True

1.21 tableNames(dbName=None)  

返回数据库的表名称列表
参数：dbName – 字符串类型的数据库名称.默认为当前的数据库。
返回: 字符串类型的表名称列表

>>> l=[(1,‘row1‘),(2,‘row2‘),(3,‘row3‘)]
>>> df = sqlContext.createDataFrame(l,[‘field1‘,‘field2‘])
>>> sqlContext.registerDataFrameAsTable(df, "table1")
>>> "table1" in sqlContext.tableNames()
True
>>> "table1" in sqlContext.tableNames("db")
True

1.22 tables(dbName=None)

返回一个包含表名称的DataFrame从给定的数据库。
如果数据库名没有指定，将使用当前的数据库。
返回的DataFrame包含两列: 表名称和是否临时表 (一个Bool类型的列，标识表是否为临时表)。

参数：● dbName – 字符串类型的使用的数据库名
返回: DataFrame

>>> l=[(1,‘row1‘),(2,‘row2‘),(3,‘row3‘)]
>>> df = sqlContext.createDataFrame(l,[‘field1‘,‘field2‘])
>>> sqlContext.registerDataFrameAsTable(df, "table1")
>>> df2 = sqlContext.tables()
>>> df2.filter("tableName = ‘table1‘").first()
Row(tableName=u‘table1‘, isTemporary=True)

1.23 udf

返回一个注册的UDF为UDFRegistration。
返回: UDFRegistration

1.24 uncacheTable(tableName)

从内存的缓存表中移除指定的表。

2.class pyspark.sql.HiveContext(sparkContext, hiveContext=None)

Hive此处暂略