- 相较于rdd，在数据挖掘中更常用的数据格式是DataFrame，由于Catalyst优化器的原因，DataFrame在python上并不比scala上慢多少
# 引入必要包
from pyspark.sql import SparkSession
from pyspark.sql import types
spark = SparkSession.builder.master("local").appName("learnsparkdf").enableHiveSupport().getOrCreate()
sc = spark.sparkContext
创建DataFrame
# 使用sc创建df
#方法一：通过json创建
stringJSONRDD = sc.parallelize(["""
{ "id": "123",
"name": "Katie",
"age": 19,
"eyeColor": "brown"
}""",
"""{
"id": "234",
"name": "Michael",
"age": 22,
"eyeColor": "green"
}""",
"""{
"id": "345",
"name": "Simone",
"age": 23,
"eyeColor": "blue"
}"""]
)
df = spark.read.json(stringJSONRDD)
df.show()
[out]:
+---+--------+---+-------+
|age|eyeColor| id| name|
+---+--------+---+-------+
| 19| brown|123| Katie|
| 22| green|234|Michael|
| 23| blue|345| Simone|
+---+--------+---+-------+
# 方法二，通过sc创建，通常要指定列名不然会变成[-1,-2]
list_rdd = sc.parallelize([('TOM', 23), ('JIM', 18), ('BOSE', 50), ('JAME',23), ('JAM')],4)
df2 = spark.createDataFrame(list_rdd)
df2.show()
print(df2.schema) # ***默认int数据类型为LongType
[out]:
+----+---+
| _1| _2|
+----+---+
| TOM| 23|
| JIM| 18|
|BOSE| 50|
|JAME| 23|
+----+---+
StructType(List(StructField(_1,StringType,true),StructField(_2,LongType,true)))
# 需要注意的是整数位LongType，与FloatType和DoubleType不能隐式转换
schema = types.StructType([
types.StructField('Name', types.StringType(), True), # 列名，数据类型，能否为空
types.StructField('Age', types.ShortType(), True),
])
df2 = spark.createDataFrame(list_rdd, schema)
df2.show()
print(df2.schema)
[out]:
+----+---+
|Name|Age|
+----+---+
| TOM| 23|
| JIM| 18|
|BOSE| 50|
|JAME| 23|
+----+---+
StructType(List(StructField(Name,StringType,true),StructField(Age,ShortType,true)))
数据类型总共有一下类型
"DataType", "NullType", "StringType", "BinaryType", "BooleanType", "DateType",
"TimestampType", "DecimalType", "DoubleType", "FloatType", "ByteType", "IntegerType",
"LongType", "ShortType", "ArrayType", "MapType", "StructField", "StructType"
创建临时表
dataframe的操作通常会使用sql语句完成，下面有四个创建表的方法
#df.createGlobalTempView("tempViewName") 创建一个全局临时表，生命周期为程序的生命周期 **使用的时候 global_temp.tempViewName
#df.createOrReplaceGlobalTempView("tempViewName") 创建或者替换一个全局临时表，生命周期为程序的生命周期
#df.createOrReplaceTempView("tempViewName") 创建一个临时表，生命周期为当前SparkSession的生命周期
#df.createTempView("tempViewName") 创建或者替换一个临时表，生命周期为当前SparkSession的生命周期
# 删除临时表
# spark.catalog.dropTempView("tempViewName")
# spark.catalog.dropGlobalTempView("tempViewName")
创建表之后，剩余的操作就和sql基本一样，一般来说sql操作都会返回一个新的dataframe
查看临时表
# 创建临时表,查看信息,
df2.createOrReplaceTempView('df')
spark.sql("select * from df").show() #***注意返回的也是一个dataframe
[out]:
+----+---+
|Name|Age|
+----+---+
| TOM| 23|
| JIM| 18|
|BOSE| 50|
|JAME| 23|
+----+---+
[Row(Name='TOM', Age=23),
Row(Name='JIM', Age=18),
Row(Name='BOSE', Age=50),
Row(Name='JAME', Age=23)]
# 取出数据后
ret= spark.sql("select * from df").collect() # 也可以直接取出来
ret[0]['name'] 类似于字典和列表综合，这两种方法都可以获取元素
ret[0][0] 类似于字典和列表综合，这两种方法都可以获取元素
数据结构
printSchema()函数
输出dataframe schema结构
若不指定dataframe结构，系统会自动推断数据类型
df.printSchema()
[out]:
root
|-- age: long (nullable = true)
|-- eyeColor: string (nullable = true)
|-- id: string (nullable = true)
|-- name: string (nullable = true)
API操作
dataframe API查询
dataframe可以通过take(),show()展示结果
可以使用select filter选择过滤数据,还有很多函数
df2.select("*").filter("age>18").show()
[out]:
+----+---+
|Name|Age|
+----+---+
| TOM| 23|
|BOSE| 50|
|JAME| 23|
+----+---+
保存文件
# 保存文件
# 一般来说可以存储为csv,json，不过更常见的是使用parquet存储
# Parquet仅仅是一种存储格式，它是语言、平台无关的，并且不需要和任何一种数据处理框架绑定，目前能够和Parquet适配的组件包括下面这些，可以看出基本上通常使用的查询引擎和计算框架都已适配，并且可以很方便的将其它序列化工具生成的数据转换成Parquet格式
df.rdd.getNumPartitions() # 获取分区数目
df.write.parquet('/FileStore/tables/testpar', mode = 'overwrite') # 以parquet格式保存数据
df.toPandas().to_csv('/FileStore/tables/testpar2.csv')# 以csv格式保存数据
from sklearn.externals import joblib
joblib.dump(df.toJSON().collect(), '/FileStore/tables/testpar3.json') # 以json格式保存
# 保存后的/FileStore/tables/testpar是一个文件加，有多少个分区数目就有多少个文件***********
df2 = spark.read.parquet('/FileStore/tables/testpar')# 读取parquet格式文件
jupyter 代码
