fe
缺点
- 不方便添加新的优化策略
- 线程安全问题
Spark SQL支持三种语言
- java
- Scala
- python
DataFrame
- 大规模数据化结构能历、提高了运算能力
- 从sql到dataFrame的转化，支持sql查询
- RDD是分布式的java对象的集合，对象颞部结构不可知
- dataframe以rdd为基础的分布式数据集，提供了详细的结构信息
DataFrame的创建
SparkSession
dataframe的常用操作
df = spark.read.json("people.json")
df.printSchema() 查看表的结构
df select(df['name'],df['age']+1).show()df.filter(df['age']>20).show()
df.groupby("age").count().show()df.sort(df['age'] desc()).show()
df.sort(df['age'].desc(),df['name'].asc()).show()利用反射机制推断RDD的模式
读取Mysql数据库中的数据
DataFrame的创建
from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession
spark = SparkSession.builder.config(conf = SparkConf()).getOrCreate()# 分布式读取文件
spark.read.text("people.txt") spark.read.format("text").load("people.txt")
spark.read.json("people.json") spark.read.format("json").load("people.json")
spark.read.parquet("people/parquet") spark.read.format("parquet").load("people.parquent")#文件保存 最后保到一个目录
df.write.txt("people.txt")
df.write.json("people.json")
df.write.parquent("people.parquent")# dataFrame的一些常用操作df.printSchema()
dat.select('_c1').show()
df.filter(df['age']>20).show()
df.groupby("age").count().show()
df.sort(df["age"].desc()).show()
df.sort(df["age"].desc(),df['name'].asc()).show()RDD转换得到dataFrame
利用反射机制推断RDD模式
#用ROW对象去封装一行一行的数据
from pyspark.sql import ROW
people = spark.sparkContext.textFile("file:///file_path").map(lambda x:x.split(",")).map(lambda x:ROW(NAME= P[0],age = int(p[1])))schemaPeople = spark.createDataFrame(people)
#必须注册为临时表才供下面的查询使用
schemaPeople.createOrReplaceTempView("people")
personDF = spark.sql("select name,age from people where age>20")
personsDRR = personsDF.rdd.map(lambda p:"Nmae"+p.name+","+age:"+str()p.age))
personsRDD.foreach(print)
