Spark各组件功能简单理解（quick start）

各个组件

• conf/spark-env.sh 配置spark的环境变量
• conf/spark-default.conf 配置spark应用默认的配置项和spark-env.sh有重合之处，可在提交应用时指定要用的配置文件（spark-submit ..... --properties-file FILE 设置应用程序属性的文件放置位置，默认是conf/spark-defaults.conf）
• bin/spark-shell 一个scala/java的spark REPL工具或者说命令行工具，可用来交互的跑一些脚本，或者测试环境之类
• bin/spark-submit 部署spark应用
• sbin/start-all.sh spark-standalone方式启动
• sbin/start-thriftserver.sh 启动thriftserver，提供了一个jdbc接口查询hive数据库

资源调度

根据spark的部署方式的不同，会有不同的组件执行资源调度（资源主要指cpu，memory）

• spark on yarn/mesos 由yarn/mesos来做资源调度
• standalone由spark自己来做资源调度

任务调度

Spark应用提交后经历了一系列的转换，最后成为Task在每个节点上执行。

1. RDD的Action算子触发Job的提交
2. 提交到Spark中的Job生成RDD DAG
3. 由DAGScheduler转化为Stage DAG
4. 每个Stage中产生相应的Task集合
5. TaskScheduler将任务分发到Executor执行
6. 每个任务对应相应的一个数据块，使用用户定义的函数处理数据
   块。

更详细内容 ： https://www.2cto.com/net/201712/703266.html

spark 基础操作

• 主要抽象是RDD(Resilient Distributed Dataset，弹性分布式数据集),它可被分发到集群各个节点上，进行并行操作。RDDs 可以通过 Hadoop InputFormats 创建（如 HDFS），或者从其他 RDDs 转化而来。
• RDD可以读取hdfs或者local的文件，可以读取text, csv, json, parquet, orc等格式的文件。
• RDD支持两种类型的操作（算子）：
  • action 在数据集上运行计算后返回值
  • transformation转换, 从现有数据集创建一个新的数据集
• transformation不会立即执行，等到一个action算子触发任务提交
• spark会根据代码生成DAG图来执行任务的调度
• rdd结合schema可以创建DataFrame，可以执行一些存表，SQL式的join，select，filter，groupby等操作。

spark-streaming 基础操作

spark的流式处理框架

• 支持kafka，flume，socket，text等方式创建Stream
• 生成的对象为DStreamRDD，通过分时生成的RDD来进行操作
• 支持常用的RDD transformation （map, flatmap,filter,repartition,union,couont,reduce等）
• 可以通过transfrom算子，将DStream作为rdd进行操作： transform(lambda rdd: func(rdd))
• 读取kafka消息有两种方式，high api式（带receiver，自动处理offset），low api式（直接操作kafka的offset），spark推荐使用low api式，自己操作offset来保证不冗余的读取，不漏读消息
• 对于python的spark-streaming，可以借助kazoo包来将offset保存到zookeeper
• 可以使用checkpoint来进行一定的容错
• 更详细可以参考：https://blog.csdn.net/weixin_35602748/article/details/78668054

pyspark-streaming demo

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

def parse_msg(spark, rdd):
    pass

if __name__ == "__main__":

    master = "yarn"
    appName = "AppName"
    kafka_params = {
        "bootstrap.servers": "master:9092",  # metadata.broker.list or bootstrap.servers
        "auto.offset.reset": "smallest"
    }

    topics = ["logger"]
    spark = SparkSession \
        .builder \
        .appName(appName) \
        .getOrCreate()

    # 优先使用sparkSession（spark2.0的方式），这样sparkContext比较容易得到，不用在自定义函数中额外传参
    sc = spark.sparkContext
    ssc = StreamingContext(sc, 10)  # 10s的数据做一次处理

    directKafkaStream = KafkaUtils.createDirectStream(ssc, topics, kafka_params)

    # get message strings from kafka
    lines = directKafkaStream.map(lambda x: x[1])

    # transform可以把DStream转换为rdd进行操作
    logJsonStrRDD = lines.transform(lambda rdd: parse_msg(spark, rdd))

    # pprint必须有，必须有一个输出，否则提示没法注册，报错： java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute
       # pprint = pretty print 用更好的格式来打印
    logJsonStrRDD.pprint(5)  

    ssc.start()
    ssc.awaitTermination()