结构化API
结构化API是处理各种数据的工具，从非结构化log文件到半结构化CSV文件和高度结构化的Parquet文件,spark中三种主要的三类结构化API为：
DatasetsDataFramesSQL tables and views
App dag stage task
DataFrame Dataset
DataFrame、 Dataset是(分布式)1-。4类表集合具有定义良好的行和列
DataFrame无类型（runtime check）
Dataset有类型 (compile time)
对于Spark来说, DataFrame是类型为 Row的Dataset
特点:
immutablelazily evaluated plans
.Scala type reference
Schemas
模式定义了一个DataFrame的列名和类型。
可以手动定义schema或者schema on read
Spark类型直接映射到Spark维护的不同语言api，在Scala、Java、Python、SQL和R中，每种api都有一个查询表，简单的说最终代码 使用纯spark执行（ Spark’s internal Catalyst representation）
结构化```API的执行流程
- 编写
DataFrame/Dataset/SQL代码. - 如果代码是正确的spark将其转化为
Logical Plan. -
Logical Plan转为Physical Plan
4.Spark在集群上执行Physical Plan(RDD操作)
Logical Planning
catalogy是一个包含所有table 和DataFrame信息的仓库，用于check代码是否有问题 （eg:table column 不存在)
check 通过的plan 通过Catalyst Optimizer优化
用户可以扩展Catalyst自定义优化规则
Physical Planning
DataFrams 相关操作
加载数据
df=spark.read.json("file:///usr/local/xldw/2015-summary.json")
或者
df = spark.read.format("json").load("file:///usr/local/xldw/2015-summary.json")
查看schema
df.schema
df.printSchema()
org.apache.spark.sql.types.StructType = ...
StructType(StructField(DEST_COUNTRY_NAME,StringType,true),
StructField(ORIGIN_COUNTRY_NAME,StringType,true),
StructField(count,LongType,true))
-
spark可以根据文件的前几行推断出schema(schema on read) -
schema是StructType实例 -
StructType由一个StructFields构成 -
Boolean代表这个列是否可以为空
手动指定Schema
// in Scala
import org.apache.spark.sql.types.{StructField, StructType, StringType, LongType}
import org.apache.spark.sql.types.Metadata
val myManualSchema = StructType(Array(StructField("DEST_COUNTRY_NAME", StringType,true),
StructField("ORIGIN_COUNTRY_NAME", StringType, true),
StructField("count", LongType, false,
Metadata.fromJson("{\"hello\":\"world\"}"))
))
val df = spark.read.format("json").schema(myManualSchema).load("/data/flight-data/json/2015-summary.json")
Columns
对spark来说, 列是一种逻辑结构，它仅表示通过表达式按每个记录计算的值. 这意味着要为列赋实值，我们需要有一行;为了得到一行，我们需要一个DataFrame
构造和引用列的两种最基本的方式：
col 或column 方法
Column作为表达式
expr函数实际上可以解析字符串中的转换和列引用，然后可以将它们传递到进一步的转换中,下面三者等价：
col("someCol") - 5
expr("someCol - 5")
expr("someCol") - 5
Columns 只是表达式
这些列和这些列的转换编译成与解析表达式相同的逻辑计划
这意味着您可以将表达式编写为DataFrame代码或SQL表达式，并获得完全相同的性能特征
from pyspark.sql.functions import expr, col, column
df.select(
expr("DEST_COUNTRY_NAME"),
col("DEST_COUNTRY_NAME"),
column("DEST_COUNTRY_NAME"))\
.show(2)
Records and Rows
spark中一行使用Row对象表示，Spark使用列表达式操作行对象，以生成可用的值，行对象在内部表示字节数组。字节数组接口永远不会显示给用户，因为我们只使用列表达式来操作它们
展示
df.show(2)
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
| United States| Romania| 15|
| United States| Croatia| 1|
+-----------------+-------------------+-----+
select and selectExpr
df.select("DEST_COUNTRY_NAME").show(2)
-- in SQL
SELECT DEST_COUNTRY_NAME FROM dfTable LIMIT 2
# select muti columns
df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)
# select function favor
df.select(expr("DEST_COUNTRY_NAME").alias("dest")).show(2)
# select sql favor
df.select(expr("DEST_COUNTRY_NAME AS destination")).show(2)
# selectExpr
df.selectExpr("DEST_COUNTRY_NAME as dest", "ORIGIN_COUNTRY_NAME").show(2)
# selectExpr opens up the true power of Spark
df.selectExpr("*",
"(DEST_COUNTRY_NAME=ORIGIN_COUNTRY_NAME) as withinCountry").show(2)
df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(1)
字面量
df.select(expr("*"), lit(1).alias("One")).show(2)
添加列
df.withColumn("numberOne", lit(1)).show(2)
-- in SQL
SELECT *, 1 as numberOne FROM dfTable LIMIT 2
df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME"))
.show(2)
//rename
df.withColumn("Destination", expr("DEST_COUNTRY_NAME")).columns
重命名列
df.withColumnRenamed("DEST_COUNTRY_NAME", "dest").columns
保留字符和关键字
withColumn不用特殊处理
selectExpr需要加上`
dfWithLongColumnName = df.withColumn("this is a long column", expr("ORIGIN_COUNTRY_NAME"))
dfWithLongColumnName.selectExpr("`this is a long column` as `long column`").show(2)
大小写敏感性
默认spark大小写不敏感，可以通过配置开启大小写敏感
set spark.sql.caseSensitive true
删除列
df.drop("ORIGIN_COUNTRY_NAME").columns
改变列的类型cast
df.withColumn("count2", col("count").cast("long"))
-- in SQL
SELECT *, cast(count as long) AS count2 FROM dfTabl
过滤行
where和filter有一样的过滤功能
df.filter(col("count") < 2).sho
df.where("count < 2").show(2)
spark自动在同一时间执行所有过滤操作，而不管过滤器的顺序如何。这意味着，如果您想指定多个过滤器，只需按顺序将它们链接起来，其余的由Spark处理
df.where(col("count") < 2).where(col("ORIGIN_COUNTRY_NAME") != "Croatia")\
.show(2)
Unique
df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()
-- in SQL
SELECT COUNT(DISTINCT(ORIGIN_COUNTRY_NAME, DEST_COUNTRY_NAME)) FROM dfTable
随机分片
dataFrames = df.randomSplit([0.25, 0.75], seed)
dataFrames[0].count() > dataFrames[1].count() # False
Union
from pyspark.sql import Row
schema = df.schema
newRows = [
Row("New Country", "Other Country", 5L),
Row("New Country 2", "Other Country 3", 1L)
]
parallelizedRows = spark.sparkContext.parallelize(newRows)
newDF = spark.createDataFrame(parallelizedRows, schema)
df.union(newDF)\
.where("count = 1")\
.where(col("ORIGIN_COUNTRY_NAME") != "United States")\
.show()
sort
sort和orderBy 有相同的排序功能
df.sort("count").show(5)
df.orderBy("count", "DEST_COUNTRY_NAME").show(5)
df.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)
from pyspark.sql.functions import desc, asc
df.orderBy(expr("count desc")).show(2)
df.orderBy(col("count").desc(), col("DEST_COUNTRY_NAME").asc()).show(2)
-- in SQL
SELECT * FROM dfTable ORDER BY count DESC, DEST_COUNTRY_NAME ASC LIMIT 2
一个高级技巧是使用asc_nulls_first、desc_nulls_first、asc_nulls_last或desc_nulls_last来指定希望DataFrame空值按顺序出现在哪里
出于优化目的，有时建议先对每个分区排序然后执行之后的transformations
spark.read.format("json").load("/data/flight-data/json/*-summary.json")\
.sortWithinPartitions("count")
Limit
df.limit(5).show()
-- in SQL
SELECT * FROM dfTable LIMIT 6
df.orderBy(expr("count desc")).limit(6).show()
-- in SQL
SELECT * FROM dfTable ORDER BY count desc LIMIT 6
重新分区和合并
Spark在driver程序中维护集群的状态。有时您需要向driver程序收集一些数据，以便在本地机器上对其进行操作
到目前为止，我们还没有明确定义这个操作。然而，我们使用了几种不同的方法来实现这一点，它们实际上都是一样的
collect从整个DataFrame获取所有数据，take获取前N行，show以表格样式打印
collectDF = df.limit(10)
collectDF.take(5) # take works with an Integer count
collectDF.show() # this prints it out nicely
collectDF.show(5, False)
collectDF.collect()
collectDF.toLocalIterator()
