spark中dataframe解析_Spark-SQL

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)

总结

以上是生活随笔为你收集整理的spark中dataframe解析_Spark-SQL的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。