(1)SparkStreaming 整合 kafka 两种方式对比
Direct 方式的优缺点分析:
南海ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景,ssl证书未来市场广阔!成为
成都创新互联的ssl证书销售渠道,可以享受市场价格4-6折优惠!如果有意向欢迎电话联系或者加微信:028-86922220(备注:SSL证书合作)期待与您的合作!
- 优点:
- 简化并行(Simplified Parallelism)。不现需要创建以及 union 多输入源,Kafka topic 的partition 与 RDD 的 partition 一一对应。
- 高效(Efficiency)。基于 Receiver-based 的方式保证数据零丢失(zero-data loss)需要配置 spark.streaming.receiver.writeAheadLog.enable=true,此种方式需要保存两份数据,浪费存储空间也影响效率。而 Direct 方式则不存在这个问题。
- 强一致语义(Exactly-once semantics)。High-level 数据由 Spark Streaming 消费,但是Offsets 则是由 Zookeeper 保存。通过参数配置,可以实现 at-least once 消费,此种情况有重复消费数据的可能。
- 降低资源。Direct 不需要 Receivers,其申请的 Executors 全部参与到计算任务中;而Receiver-based 则需要专门的 Receivers 来读取 Kafka 数据且不参与计算。因此相同的资源申请,Direct 能够支持更大的业务。
- 降低内存。Receiver-based 的 Receiver 与其他 Exectuor 是异步的,并持续不断接收数据,对于小业务量的场景还好,如果遇到大业务量时,需要提高 Receiver 的内存,但是参与计算的 Executor 并无需那么多的内存。而 Direct 因为没有 Receiver,而是在计算时读取数据,然后直接计算,所以对内存的要求很低。
- 缺点:
- 提高成本。Direct 需要用户采用 checkpoint 或者第三方存储来维护 offsets,而不像Receiver-based 那样,通过 ZooKeeper 来维护 Offsets,此提高了用户的开发成本。
- 监控可视化。Receiver-based 方式指定 topic 指定 consumer 的消费情况均能通过ZooKeeper 来监控,而 Direct 则没有这种便利,不能自动保存 offset 到 zookeeper,如果做到监控并可视化,则需要投入人力开发。
Receiver 方式的优缺点分析:
- 优点:
- 专注计算。Kafka 的 high-level 数据读取方式让用户可以专注于所读数据,而不用关注或维护 consumer 的 offsets,这减少用户的工作量以及代码量而且相对比较简单。
- 缺点:
- 防数据丢失。做 checkpoint 操作以及配置 spark.streaming.receiver.writeAheadLog.enable参数,配置 spark.streaming.receiver.writeAheadLog.enable 参数,每次处理之前需要将该batch 内的日志备份到 checkpoint 目录中,这降低了数据处理效率,反过来又加重了Receiver 端的压力;另外由于数据备份机制,会受到负载影响,负载一高就会出现延迟的风险,导致应用崩溃。
- 单 Receiver 内存。由于 receiver 也是属于 Executor 的一部分,那么为了提高吞吐量
- 重复消费。在程序失败恢复时,有可能出现数据部分落地,但是程序失败,未更新 offset的情况,这导致数据重复消费。
- Receiver 和计算的 Executor的异步的,那么遇到网络等因素原因,导致计算出现延迟,计算队列一直在增加,而Receiver 则在一直接收数据,这非常容易导致程序崩溃。
(2)对kafka消费的offset的管理
- spark自带的checkpoint:
- 启用spark streaming的checkpoint是存储偏移量的最简单方法
- 流式checkpoint专门保存用户应用程序的状态
- 但是checkpoint的目录是不能共享的,无法跨越应用程序进行恢复
- 一般不使用checkpoint管理offset
- 使用zookeeper管理offset
- 如果Zookeeper中未保存offset,根据kafkaParam的配置使用最新或者最旧的offset
- 如果 zookeeper中有保存offset,我们会利用这个offset作为kafkaStream 的起始位置
- 使用hbase保存offset
- Rowkey的设计:topic名称 + groupid + streaming的batchtime.milliSeconds
- 使用hdfs管理offset:当然这种情况不推荐使用,因为在hdfs中会生成大量的小文件,导致,hdfs的性能急剧下降
(3)Driver的HA
介绍:他能够在driver失败的时候,通过读取checkpoint目录下的元数据,恢复当前streamingContext对象的状态;它能够察觉到driver进程异常退出之后,自动重启。
具体流程:当第一次运行程序时,发现checkpoint中没有数据,则根据定义的函数来第一次创建StreamingContext对象,当程序异常退出的时候,此时会根据checkpoint中的元数据恢复一个StreamingContext对象,达到异常退出之前的状态,而实现异常退出并自动启动则是sparkStreaming应用程序对driver进行监控,并且在他失败的时候感知,并进行重启。
必要条件:
- spark-submit提交作业的时候,必须是集群模式(cluster),并且必须在spark-standalong下。
spark-submit \
--class com.aura.mazh.spark.streaming.kafka.SparkStreamDemo_Direct \
//这里只能使用spark的standalong模式,所以配置为spark集群
--master spark://hadoop02:7077,hadoop04:7077 \
--driver-memory 512m \
--total-executor-cores 3 \
--executor-memory 512m \
#这句代码一定要加,他可以使异常退出的driver程序,重新启动
--supervise \
--name SparkStreamDemo_Direct \
--jars /home/hadoop/lib/kafka_2.11-0.8.2.1.jar,\
/home/hadoop/lib/metrics-core-2.2.0.jar,\
/home/hadoop/lib/spark-streaming_2.11-2.3.2.jar,\
/home/hadoop/lib/spark-streaming-kafka-0-8_2.11-2.3.2.jar,\
/home/hadoop/lib/zkclient-0.3.jar \
/home/hadoop/original-spark-1.0-SNAPSHOT.jar \
spark://hadoop02:7077,hadoop04:7077
- 需要添加--supervise \,才能实现失败自启动
- 需要配置checkpoint目录,并且是存储在hdfs上,jar也要放置在hdfs上
另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。
标题名称:SparkStreaming整合kafka的补充-创新互联
路径分享:
http://bjjierui.cn/article/ccpsoj.html