摘要:在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''), name=u'jack'), Row(addr=Row(city=u'shanghai', mail=u''), name=u'john')]
>>> df2 = sqlContext.jsonRDD(json,df1.schema)
>>> df2.collect()
[Row(addr=Row(city=u'beijing', mail=u''), name=u'jack'), Row(addr=Row(city=u'shanghai', mail=u''), 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'')]
>>> 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此处暂略

《Spark Python API 官方文档中文版》 之 pyspark.sql (一)的更多相关文章

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

    摘要:在Spark开发中,由于需要用Python实现,发现API与Scala的略有不同,而Python API的中文资料相对很少.每次去查英文版API的说明相对比较慢,还是中文版比较容易get到所需, ...

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

    摘要:在Spark开发中,由于需要用Python实现,发现API与Scala的略有不同,而Python API的中文资料相对很少.每次去查英文版API的说明相对比较慢,还是中文版比较容易get到所需, ...

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

    摘要:在Spark开发中,由于需要用Python实现,发现API与Scala的略有不同,而Python API的中文资料相对很少.每次去查英文版API的说明相对比较慢,还是中文版比较容易get到所需, ...

  4. 人工智能系统Google开源的TensorFlow官方文档中文版

    人工智能系统Google开源的TensorFlow官方文档中文版 2015年11月9日,Google发布人工智能系统TensorFlow并宣布开源,机器学习作为人工智能的一种类型,可以让软件根据大量的 ...

  5. TestNG官方文档中文版(2)-annotation(转)

    1. 介绍    TestNG是一个设计用来简化广泛的测试需求的测试框架,从单元测试(隔离测试一个类)到集成测试(测试由有多个类多个包甚至多个外部框架组成的整个系统,例如运用服务器). 编写一个测试的 ...

  6. Istio官方文档中文版

    Istio官方文档中文版 http://istio.doczh.cn/ https://istio.io/docs/concepts/what-is-istio/goals.html 为什么要使用Is ...

  7. TensorFlow 官方文档中文版

    http://wiki.jikexueyuan.com/list/deep-learning/ TensorFlow 官方文档中文版 你正在阅读的项目可能会比 Android 系统更加深远地影响着世界 ...

  8. 2DToolkit官方文档中文版打地鼠教程(三):Sprite Collections 精灵集合

    这是2DToolkit官方文档中 Whack a Mole 打地鼠教程的译文,为了减少文中过多重复操作的翻译,以及一些无必要的句子,这里我假设你有Unity的基础知识(例如了解如何新建Sprite等) ...

  9. 2DToolkit官方文档中文版打地鼠教程(二):设置摄像机

    这是2DToolkit官方文档中 Whack a Mole 打地鼠教程的译文,为了减少文中过多重复操作的翻译,以及一些无必要的句子,这里我假设你有Unity的基础知识(例如了解如何新建Sprite等) ...

随机推荐

  1. 《C#图解教程》读书笔记之三:方法

    本篇已收录至<C#图解教程>读书笔记目录贴,点击访问该目录可获取更多内容. 一.方法那些事儿 (1)方法的结构:方法头—指定方法的特征,方法体—可执行代码的语句序列: (2)方法的调用:参 ...

  2. ZOJ 3593 One Person Game

    One Person Game Time Limit: 2 Seconds      Memory Limit: 65536 KB There is an interesting and simple ...

  3. Oracle-单表合并列

    表基本结构 合并列 select t.student,decode(t.java,'','','java') 科目, t.java from student t union select t.stud ...

  4. (转)Java通过axis调用WebService

    转自:http://blog.csdn.net/wanglha/article/details/49679825 转载地址:http://www.linuxidc.com/Linux/2015-06/ ...

  5. Class类对象的三种实例化方法

    class X { } public class GetClassDemo01 {     public static void main(String[] args) {         X x = ...

  6. 企业架构研究总结(39)——TOGAF架构能力框架之架构委员会和架构合规性

    3. 架构委员会 正如前面所说,一个用来对架构治理策略的实现进行监督的跨组织的架构委员会是架构治理策略成功的主要要素之一.架构委员会应该能够代表所有主要干系人的需求,并且通常还需要对整个架构的审查及维 ...

  7. scrapy配置

    scrapy配置 增加并发 并发是指同时处理的request的数量.其有全局限制和局部(每个网站)的限制. Scrapy默认的全局并发限制对同时爬取大量网站的情况并不适用,因此您需要增加这个值. 增加 ...

  8. 字符、字符集、编码,以及它们python中会遇到的一些问题(下)

    在看了很多的博客文章之后,总结整理得到了以下文章,非常感谢这些无私奉献的博主! 文章末尾有本文引用的文章的链接,如果有漏掉的文章引用,可以发邮件联系我,随后再次附上链接! 侵删!!! 这一部分是下篇, ...

  9. 面试题-浅谈JavaScript中的This指向问题

    各位小伙伴在面试中被面试官问道this指向问题一定不少吧,同时还被问道apply,call和bind的用法区别,现在,就来简单的聊一聊this到底指向何方. 1.基本概念 MDN的官方解释:与其他语言 ...

  10. css新单位vw,vh在响应式设计中的应用

    考虑到未来响应式设计的开发,如果你需要,浏览器的高度也可以基于百分比值调整.但使用基于百分比值并不总是相对于浏览器窗口的大小定义的最佳方式,比如字体大小不会随着你窗口改变而改变,如今css3引入的新单 ...