spark rdd
spark sql
一个用于处理结构化数据的spark组件，强调的是“结构化数据”，而非“SQL”
Spark SQL不仅仅是SQL这么简单的事情，它还能做更多的事情
Hive：SQL
Spark SQL：SQL
Spark SQL提供的操作数据方式
SQL
DataFrame API
Dataset API
Spark SQL架构
Datasets and DataFrames
Spark RDD vs MapReduce
R/Pandas ： one machine
==> DataFrame：让小伙伴们感觉开发单机版应用程序一样来开发分布式应用程序
spark 1.6版本增加Dataset，只支持scala和java
DataFrame：以列的形式构成分布式的数据集
面试题：RDD与DataFrame区别
编程
from pyspark.sq import SparkSession
def basic(spark):
df = spark.read.json("test.json")
df.show()
df.printSchema()
# 选择一个字段
df.select("name").show()
# 处理
df.select(df["name"],df["age"] + 1).show()
# filter
df.filter(df['age'] > 21).show()
# groupy
df.groupBy('age').count().show()
# register temporary view
df.createOrReplaceTempView("people")
sqlDF = spark.sql("select * from people")
sqlDF.show()
spark = SparkSession.builder.appName("test").getOrCreate()
basic(spark)
spark.stop()
spark core 总结
spark core是spark生态最核心的部分
spark生态
spark core mapreduce
spark sql 类似于hive
spark streaming 类似storm，flink
spark ML 基于dataframe sparkmllib rdd
spark
基于内存的分布式计算框架
mapreduce 和 spark 优劣
spark 基于内存 计算快
spark api 更丰富，比mapreduce代码少
spark生态完整 （离线计算spark core/sql，实时计算/流式计算spark streaming，交互计算spark sql，机器学习spark ml）
RDD
弹性分布式数据集
不可变 （rdd1 -> rdd2 两个状态分别保存）
弹性 （存储弹性，分布式弹性，容错可以分成多个partition，每个partition有多个副本）
分布式
并行计算
RDD创建
spark context（在内存中 list iterable；文件中加载；创建rdd可以指定partition的数量，一个partiton对应一个task）
三类算子
transformation （返回rdd，延迟执行）
action （获取结果）
spark 模式
local
standalone
广播变量
spark sql
spark中用于处理结构化数据的一个模块（功能和hive类似）
优势
代码少，计算快（对比直接写rdd，转换成rdd会做执行优化，比自己写的效率更高；rdd缺点：序列化和反序列化开销大，频繁创建和销毁对象造成大量GC）
spark sql dataframe
DataFrame 是一个分布式的行集合
基于RDD（不可修改，延迟执行，分布式）
DataFrame vs RDD
DataFrame相当于带着schema的RDD
DataFrame引入了off-heap,意味着jvm堆以外的内存
RDD是分布式java对象集合，DataFrame是分布式Row对象集合
pandas vs spark
单机 vs 分布式计算
spark DataFrame 延迟执行，不可变
pandas DataFrame api 更丰富
Spark Streaming
实时计算流式计算框架
storm（来一条处理一条，比spark streaming快）
flink（来一条处理一条，比spark streaming快）
spark streaming （micro batch；间隔一段时间取一次数据；能处理的数据量会更大，实时性稍差）
spark streaming 组件
Streaming Context
流上下文通过Streaming Context 可以连接数据源获取数据
通过spark context 可以获取Streaming Context
在创建Streaming Context 需要指定一个时间间隔（micro batch）
Streaming Context 调用stop方法之后就不能再次调start（）
一个SparkContext创建一个Streaming Context
Streaming Context调用stop方法，默认把spark context关掉
对DStream中的数据处理逻辑要写在Streaming Context之前，一旦Streaming Context调用start方法就不能添加新的数据处理逻辑
DStream
streaming context 连接到不同的数据源获取数据，抽象成DStream
代表一个连续的数据流
由一系列连续的RDD组成
案例：updateStateByKey
from pyspark.streaming import StreamingContext
from pysql.sql.session import SparkSession
#创建sparkContext
spark = SparkSession.builder.master("local[2]").getOrCreate()
sc = spark.sparkContext
ssc = StreamingContext(sc,3)
#开启检查点
ssc.checkpoint("checkpoint")
#定义state更新函数
def updateFunc(new_values,last_sum):
return sum(new_values) + (last_sum or 0)
# 对数据进行拆分，分为多个单词
lines = ssc.socketTextStream("localhost",9999)
counts = lines.flapMap(lambda:line:line.split(" ")).map(lambda word:(word,1)).updateStateByKey(updateFunc=updateFunc)
counts.pprint()
ssc.start()
ssc.awaitTermination()
windows
窗口长度L：运算数据量
滑动间隔G：控制间隔
from pyspark import SparkContext
from pyspark.steaming import StreamingContext
from pyspark.sql.session import SparkSession
def get_countryname(line):
country_name = line.strip()
if country_name = 'usa':
output = 'USA'
if country_name = 'ind':
output = 'India'
else:
output = 'UnKnow'
if __name__ == "__main__":
batch_interval = 1
window_length = 6 * batch_interval
frequency = 3 * batch_interval
#创建sparkContext
spark = SparkSession.builder.master("local[2]").getOrCreate()
sc = spark.sparkContext
ssc = StreamingContext(sc,batch_interval)
#开启检查点
ssc.checkpoint("checkpoint")
addFunc = lambda:x,y:x+y
invAddFunc = lambda :x,y:x-y
window_counts = lines.map(get_countryname).reduceByKeyAndWindow(andFunc,invAddFunc,window_length,frequency)
window_counts.pprint()
ssc.start()
ssc.awaitTermination()
