1.pandas df 与 spark df的相互转换

import pandas as pd
import numpy as np
arr = np.arange(9).reshape(-1,3)
df_p = pd.DataFrame(arr)
arr
df_p
df_p.columns=['a','b','c']
df_p
df_s = spark.createDataFrame(df_p)
df_s.show()
df_s.collect()
df_s.show()
df_s.toPandas()

2. Spark与Pandas中DataFrame对比

http://www.lining0806.com/spark与pandas中dataframe对比/

3.1 利用反射机制推断RDD模式

sc创建RDD
转换成Row元素，列名=值
spark.createDataFrame生成df
df.show(), df.printSchema()

from pyspark.sql import Row
people = spark.sparkContext.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt").map(lambda line:line.split(',')).map(lambda p:Row(name=p[0],age=int(p[1])))
schemaPeople = spark.createDataFrame(people)
schemaPeople.createOrReplaceTempView("people")
personsDF = spark.sql("select name,age from people where age>20")
personsRDD = personsDF.rdd.map(lambda p:"Name:"+p.name+","+"Age:"+str(p.age))
personsRDD.foreach(print)
schemaPeople.show()
schemaPeople.printSchema()

3.2 使用编程方式定义RDD模式

生成“表头”
fields = [StructField(field_name, StringType(), True) ,...]
schema = StructType(fields)
生成“表中的记录”
创建RDD
转换成Row元素，列名=值
把“表头”和“表中的记录”拼装在一起
= spark.createDataFrame(RDD, schema)

from pyspark.sql.types import StringType,StructField,StructType
from pyspark.sql import Row
schemaString = "name age"
fields = [StructField(field_name,StringType(),True) for field_name in schemaString.split(" ")]
schema = StructType(fields)
fields
schema
lines = spark.sparkContext.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt")
parts = lines.map(lambda x:x.split(","))
people = parts.map(lambda p:Row(p[0],p[1].strip()))
people.collect()
schemaPeople = spark.createDataFrame(people,schema)
schemaPeople.show()
schemaPeople.printSchema()

4. DataFrame保存为文件

df.write.json(dir)

schemaPeople.write.json("file:///usr/local/spark/schema_out")