pyspark基础 -- DataFrame的理解与案例

DataFrame(df)介绍

datafram就是一个内存中的二维表结构，具备表结构的三个基本属性：

行
列
表结构描述

在结构层面，pyspark中的StructType对象描述了表结构，StructField对象描述了表的一个列信息；
在数据层面，Row对象记录了一行数据，Column对象记录了一列数据并包含列信息；

在这里插入图片描述

DataFrame对象的构建方式

方式一基于RDD构建

dataframe是可以从RDD中直接转化而来的，通过sparksession对象的createDataFrame方法可以将RDD转化为DataFrame,下例：
假设我们有一个people.txt，内容如下：

zhangsan,28
lisi,29
wangwu,30

# coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':# 构建sparkspark = SparkSession.builder.\appName("Create df").\master("local[*]").\getOrCreate()# 获取contextsc = spark.sparkContext# 构建一个RDD rdd[(name, age), ()]rdd = sc.textFile("people.txt").\map(lambda x : x.split(',')).\map(lambda x : [x[0], int(x[1])]) # RDD会做类型自动探测，这里需要做类型转换# 构建df，RDD按照提供的字段顺序一次获取信息df = spark.createDataFrame(rdd, schema = ["name", "age"])# 打印df结构df.printSchema()# 打印20行数据# show有两个参数，参数1指定展示多少条数据，默认20# 参数2表示是否对列进行截断，如果列的长度超过20个字符传长度，后续的内容以...代替，不全打印df.show()# df.show(15, False)# 构建临时视图表，让我们可以用sparksql的方式查询表内容df.createTempView("peopleTable")spark.sql("select * from peopleTable where age < 29").show()

方式二通过StructType对象构建

structtype类可以定义整个DataFrame中的schema，也即 df = spark.createDataFrame(rdd, schema = [“name”, “age”])这里的schema可以通过structType来指定

# 需导入以下类
from pyspark.sql import SturctType StringType IntegerType
# 定义表结构，第一个为列名称，第二个参数列数据类型，第三个是否允许为空
schema = StructTpe.\add("id", IntegerType(), nullable=False).\add("name", StringType(), nullable=False).\add("score", IntegerType(), nullable=False)
# 将rdd转为scheme结构
df = spark.createDataFrame(rdd, schema)

通过rdd.toDF()的方法

rdd对象本身有toDF()方法，可以通过参数指定表结构来将rdd转为DataFrame，参数也细分为两种，一种是直接指定表头，如下：

df1 = rdd.toDF(["name", "age"])

这种方法只能指定列名，无法指定列数据类型，只能靠编译器自动推断类型，在类型不敏感时可用，另外一种则是利用上面提到的StructType类先定义号表结构，传入指定表结构变量

df2 = rdd.toDF(schema=schema)

基于pandas的DataFrame转化为sparksql的DataFrame

createDataFrame直接接收pandas的df即可转换

from pyspark.sql import SparkSession
import pandas as pdif __name__ == '__main__':spark = SparkSession.builder.\appName("test").\master("local[*]").\getOrCreate()sc = spark.sparkContext# 定义pandas的DataFramepdf = pd.DataFrame({"id" : [1, 2, 5],"name" : ["zhang", "wang", "li"],"age" : [11, 23, 55]})# 转换为spark的DFdf = spark.createDataFrame(pdf)