spark相关

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()