Spark2.1.0+入门:从RDD转换得到DataFrame(Python版)

大数据学习路线图

【版权声明】博客内容由厦门大学数据库实验室拥有版权,未经允许,请勿转载!
返回Spark教程首页
推荐纸质教材:林子雨、郑海山、赖永炫编著《Spark编程基础(Python版)》

Spark官网提供了两种方法来实现从RDD转换得到DataFrame,第一种方法是,利用反射来推断包含特定类型对象的RDD的schema,适用对已知数据结构的RDD转换;第二种方法是,使用编程接口,构造一个schema并将其应用在已知的RDD上。

利用反射机制推断RDD模式

在利用反射机制推断RDD模式时,我们会用到toDF()方法
下面是在pyspark中执行命令以及反馈的信息:

>>> from pyspark.sql.types import Row
>>> def f(x):
...     rel = {}
...     rel['name'] = x[0]
...     rel['age'] = x[1]
...     return rel
... 
>>> peopleDF = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt").map(lambda line : line.split(',')).map(lambda x: Row(**f(x))).toDF()
>>> peopleDF.createOrReplaceTempView("people")  //必须注册为临时表才能供下面的查询使用

>>> personsDF = spark.sql("select * from people")
>>> personsDF.rdd.map(lambda t : "Name:"+t[0]+","+"Age:"+t[1]).foreach(print)

Name: 19,Age:Justin
Name: 29,Age:Michael
Name: 30,Age:Andy

使用编程方式定义RDD模式

使用createDataFrame(rdd, schema)编程方式定义RDD模式。

>>>  from pyspark.sql.types import Row
>>>  from pyspark.sql.types import StructType
>>> from pyspark.sql.types import StructField
>>> from pyspark.sql.types import StringType

//生成 RDD
>>> peopleRDD = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt")

//定义一个模式字符串
>>> schemaString = "name age"

//根据模式字符串生成模式
>>> fields = list(map( lambda fieldName : StructField(fieldName, StringType(), nullable = True), schemaString.split(" ")))
>>> schema = StructType(fields)
//从上面信息可以看出,schema描述了模式信息,模式中包含name和age两个字段


>>> rowRDD = peopleRDD.map(lambda line : line.split(',')).map(lambda attributes : Row(attributes[0], attributes[1]))

>>> peopleDF = spark.createDataFrame(rowRDD, schema)

//必须注册为临时表才能供下面查询使用
scala> peopleDF.createOrReplaceTempView("people")

>>> results = spark.sql("SELECT * FROM people")
>>> results.rdd.map( lambda attributes : "name: " + attributes[0]+","+"age:"+attributes[1]).foreach(print)

name: Michael,age: 29
name: Andy,age: 30
name: Justin,age: 19

在上面的代码中,peopleRDD.map(lambda line : line.split(‘,’))作用是对people这个RDD中的每一行元素都进行解析。比如,people这个RDD的第一行是:

Michael, 29

这行内容经过peopleRDD.map(lambda line : line.split(‘,’)).操作后,就得到一个集合{Michael,29}。后面经过map(lambda attributes : Row(attributes[0], attributes[1]))操作时,这时的p就是这个集合{Michael,29},这时p[0]就是Micheael,p[1]就是29,map(lambda attributes : Row(attributes[0], attributes[1]))就会生成一个Row对象,这个对象里面包含了两个字段的值,这个Row对象就构成了rowRDD中的其中一个元素。因为people有3行文本,所以,最终,rowRDD中会包含3个元素,每个元素都是org.apache.spark.sql.Row类型。实际上,Row对象只是对基本数据类型(比如整型或字符串)的数组的封装,本质就是一个定长的字段数组。
peopleDF = spark.createDataFrame(rowRDD, schema),这条语句就相当于建立了rowRDD数据集和模式之间的对应关系,从而我们就知道对于rowRDD的每行记录,第一个字段的名称是schema中的“name”,第二个字段的名称是schema中的“age”。

把RDD保存成文件

这里介绍如何把RDD保存成文本文件,后面还会介绍其他格式的保存。

第1种保存方法

进入pyspark执行下面命令:

>>> peopleDF = spark.read.format("json").load("file:///usr/local/spark/examples/src/main/resources/people.json")

>>> peopleDF.select("name", "age").write.format("csv").save("file:///usr/local/spark/mycode/newpeople.csv")

可以看出,这里使用select(“name”, “age”)确定要把哪些列进行保存,然后调用write.format(“csv”).save ()保存成csv文件。在后面小节中,我们还会介绍其他保存方式。
另外,write.format()支持输出 json,parquet, jdbc, orc, libsvm, csv, text等格式文件,如果要输出文本文件,可以采用write.format(“text”),但是,需要注意,只有select()中只存在一个列时,才允许保存成文本文件,如果存在两个列,比如select(“name”, “age”),就不能保存成文本文件。

上述过程执行结束后,可以打开第二个终端窗口,在Shell命令提示符下查看新生成的newpeople.csv:

cd  /usr/local/spark/mycode/
ls

可以看到/usr/local/spark/mycode/这个目录下面有个newpeople.csv文件夹(注意,不是文件),这个文件夹中包含下面两个文件:

part-r-00000-33184449-cb15-454c-a30f-9bb43faccac1.csv 
_SUCCESS

不用理会_SUCCESS这个文件,只要看一下part-r-00000-33184449-cb15-454c-a30f-9bb43faccac1.csv这个文件,可以用vim编辑器打开这个文件查看它的内容,该文件内容如下:

Michael,
Andy,30
Justin,19

因为people.json文件中,Michael这个名字不存在对应的age,所以,上面第一行逗号后面没有内容。
如果我们要再次把newpeople.csv中的数据加载到RDD中,可以直接使用newpeople.csv目录名称,而不需要使用part-r-00000-33184449-cb15-454c-a30f-9bb43faccac1.csv 文件,如下:

>>> textFile = sc.textFile("file:///usr/local/spark/mycode/newpeople.csv")
>>> textFile.foreach(print)
Justin,19
Michael,
Andy,30

第2种保存方法

进入pyspark执行下面命令:

>>> peopleDF = spark.read.format("json").load("file:///usr/local/spark/examples/src/main/resources/people.json"
>>> peopleDF.rdd.saveAsTextFile("file:///usr/local/spark/mycode/newpeople.txt")

可以看出,我们是把DataFrame转换成RDD,然后调用saveAsTextFile()保存成文本文件。在后面小节中,我们还会介绍其他保存方式。
上述过程执行结束后,可以打开第二个终端窗口,在Shell命令提示符下查看新生成的newpeople.txt:

cd  /usr/local/spark/mycode/
ls

可以看到/usr/local/spark/mycode/这个目录下面有个newpeople.txt文件夹(注意,不是文件),这个文件夹中包含下面两个文件:

part-00000  
_SUCCESS

不用理会_SUCCESS这个文件,只要看一下part-00000这个文件,可以用vim编辑器打开这个文件查看它的内容,该文件内容如下:

[null,Michael]
[30,Andy]
[19,Justin]

如果我们要再次把newpeople.txt中的数据加载到RDD中,可以直接使用newpeople.txt目录名称,而不需要使用part-00000文件,如下:

>>> textFile = sc.textFile("file:///usr/local/spark/mycode/newpeople.txt")
>>> textFile.foreach(print)
[null,Michael]
[30,Andy]
[19,Justin]