各个组件
- 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在每个节点上执行。
- RDD的Action算子触发Job的提交
- 提交到Spark中的Job生成RDD DAG
- 由DAGScheduler转化为Stage DAG
- 每个Stage中产生相应的Task集合
- TaskScheduler将任务分发到Executor执行
- 每个任务对应相应的一个数据块,使用用户定义的函数处理数据
块。
更详细内容 : 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()
网友评论