14-pyspark的DataFrame使用总结

10-用PySpark建立第一个Spark RDD(PySpark实战笔记系列第一篇)
11-pyspark的RDD的变换与动作算子总结(PySpark实战笔记系列第二篇))
12-pyspark的RDD算子注意事项总结(PySpark实战笔记系列第三篇)
13-pyspark的共享变量用法总结(PySpark实战笔记系列第四篇)
14-pyspark的DataFrame使用总结(PySpark实战笔记系列第五篇)

前言

在Spark中，除了RDD这种数据容器外，另一种一种更容易操作的一个分布式数据容器DataFrame，它更像传统关系型数据库的二维表，除了包括数据自身以外，还包括数据的结构信息（Schema），可以利用类似SQL的语言来进行数据访问。

DataFrame可以从多种数据来源上进行构建，比如结构化数据文件、Hive中的表、外部数据库或现有RDD。

DataFrame使用总结

DataFrame的构建

方法1：通过列表构建

列表的元素是元组，这个数据结构可以代表一种二维数据。然后利用spark.createDataFrame()方法来构建，示例如下：

import findspark
findspark.init()#############################################from pyspark.sql
import SparkSession
spark = SparkSession.builder \
                    .master("local[2]") \
                    .appName("DataFrameDemo") \
                    .getOrCreate();############################################
a =[('Jack',32),('Smith',33)]
df = spark.createDataFrame(a)#[Row(_1='Jack', _2=32), Row(_1='Smith', _2=33)]print(df.collect())
df.show()15# +-----+---+  # |   _1| _2|# +-----+---+# | Jack| 32|# |Smith| 33|# +-----+---+# 指定列名
df2 = spark.createDataFrame(a,['name','age'])#[Row(name='Jack', age=32), Row(name='Smith', age=33)]print(df2.collect())
df2.show()# +-----+---+# | name|age|# +-----+---+# | Jack| 32|# |Smith| 33|3# +-----+---+

方法2：通过Row对象构建

到DataFrame对象是由Row这个数据结构构成的，因此也可以用Row，然后利用**spark.createDataFrame() 方法 **来创建DataFrame对象。示例如下：

# 通用的开头# ......#################################################from pyspark.sql import Row 

a =[('Jk',32),('Sm',33)]
rdd = sc.parallelize(a)# 创建包含列名的Row
RMes= Row('name','age')# rdd对象的元素进行映射，转换成一个RMes对象，并返回一个新RDD对象
rmes = rdd.map(lambda r: RMes(*r))
df = spark.createDataFrame(rmes)# [Row(name='Jk', age=32), Row(name='Sm', age=33)]print(df.collect())
df.show()# +-----+---+# | name|age|# +-----+---+# | Jk  | 32|# |Sm   | 33|# +-----+---+

方法3：通过表Schema构建

上述两个方法都没能给定每个字段的类型，比如列名name是字符串类型，而列名age是数值类型。而通过用StructType方法创建了一个表Schema则可以实现，类似定义数据库中的表结构。再利用spark.createDataFrame()方法来创建DataFrame对象。示例如下：

# 通用的开头# ......#################################################from pyspark.sql.types import*

a =[('Jk',32),('Sm',33)]
rdd = sc.parallelize(a)# 用StructType方法创建了一个表Schema
schema = StructType([
            StructField("name", StringType(),True),
            StructField("age", IntegerType(),True)])# 创建DataFrame
df = spark.createDataFrame(rdd, schema)# Row(name='Jk', age=32), Row(name='Sm', age=33)]print(df.collect())
df.show()# +-----+---+# | name|age|# +-----+---+# | Jk  | 32|# |Sm   | 33|# +-----+---+
df.printSchema()# root#  |-- name: string (nullable = true)#  |-- age: integer (nullable = true)

方法4：rdd结合字符串构建

借助StructType方法可以创建类型化的DataFrame对象，但是操作起来有点繁琐。下面示例一个简单一点的方法，同样可以创建具备字段类型的DataFrame对象。

# 通用的开头# ......#################################################
a =[('Jk',32),('Sm',33)]
rdd = sc.parallelize(a)# 创建DataFrame:使用一个字符串对表结构中的字段类型进行定义
df = spark.createDataFrame(rdd,"name:string, age:int")# Row(name='Jk', age=32), Row(name='Sm', age=33)]print(df.collect())
df.show()# +-----+---+# | name|age|# +-----+---+# | Jk  | 32|# |Sm   | 33|# +-----+---+
df.printSchema()# root#  |-- name: string (nullable = true)#  |-- age: integer (nullable = true)