“五招” 迅速提升你的流处理程序

今天解读的内容是来自 Spark Summit East 2017 关于 Spark Streaming 应用程序开发的一个介绍：What No One Tells You About Writing a Streaming App。

注：本文只是解读此技术分享的要点，如果对具体内容有兴趣，请查看 PPT 或视频，资料下载请关注微信公众号：大数据技术峰会解读（bigdata_summit），回复 “001”

作者主要针对 Spark Streaming 流处理程序的如下五个方面进行探讨：

• 监控和管理流处理作业
• 如何防止数据丢失
• 什么场景使用 Spark Streaming
• Exact Once 语义
• 如何优雅地关闭流处理程序

Monitoring and managing jobs（ 监控和管理作业）

主要是针对运行在 YARN 之上的 spark streaming 流处理程序进行探讨并回答如下方面的问题：

作业管理：

• 在哪里运行 driver 程序？
  • 以 yarn cluster 模型部署运行，driver 运行在 YARN 中
• 当 driver 程序失败的时候，如何自动地重启 driver 程序？
  • 设置 spark.yarn.maxAppAttempts 和 spark.yarn.am.attemptFailuresValidityInterval
• 如何暂停作业？
  • 参考 “如何优雅地关闭流处理程序”

作业监控：

• 如何防止待处理数据堆积？
• 如何监控长时间运行的 driver 程序的健康状态？

Prevent data loss （防止数据丢失）

由上面可知，当 Driver 程序宕掉后，可以自动重启，但是这过程中如何防止数据丢失呢？不同的情况需要不同对待：

1. File Sources（当数据源来自文件，如 HDFS 文件或 S3）

• 主要是使用 Checkpoint

2. Reciever based sources（有专门的 executor 接受来自外部 [如 socket 等] 的数据，供后续其他 executor 处理）

• Checkpoint + WAL（Write Ahead Log）

3. Spark Streaming with Kafka

• 使用 Direct Connect 模式，无需 WAL
• 但是应用程序需要记录 kafka offset，一种做法是使用 checkpoint，但是作者不建议这么做，因为一旦更新应用程序，老程序的 checkpoint 就无法使用了，建议使用 Zookeeper，HDFS 或者数据库来记录 kafka offset。

扩展阅读：

• Spark streaming 与 kafka 0.8 集成： http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html#approach-2-direct-approach-no-receivers
• Spark streaming 与 kafka 0.10 集成： http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
• Improved Fault-tolerance and Zero Data Loss in Apache Spark Streaming: https://databricks.com/blog/2015/01/15/improved-driver-fault-tolerance-and-zero-data-loss-in-spark-streaming.html
• Tracking offsets when using direct connector: http://blog.cloudera.com/blog/2015/03/exactly-once-spark-streaming-from-apache-kafka/

Do I really need to use Spark Streaming （真的需要使用 Spark Streaming 吗？）

Spark Streaming 比较适合如下场景：

• Accutate Counts（精确地计数）
• Windows aggregations（基于时间窗口的聚合运算）
• Progressive Analysis
• Continuous Machine Learning（连续第做 ML 计算，如根据结果持续地优化 ML 模型）

如下方面 Spark Streaming 也能做，但是作者认为 Spark Streaming 未必是最好的选择

• Automic Enrichment
• Notifications
• Joining
• Partitioning
• Ingestion

Exactly once sematics （Exactly once 语义）

作者认为，虽然 spark 引擎能提供 Exactly Once 语义，但是它很可能只是整个流水线的一（小）部分，任何一个其他地方（如数据源、Sink 等）出错，可能从应用层面就无法保证计算的完全精确。

Graceful shutting down your streaming app （如何优雅地关闭流处理程序）

首先定义如何关闭流处理程序才算是 “优雅地关闭”：

• Offset known （这里的 offset 指的应该是 Kafka 的 offset）
• State stored externally （比如 checkpoint）
• Stopping at the right place（即等待当前的 micro-batch 处理完成的时候）

介绍了两种优雅关闭Spark Streaming 程序的方法：

• 使用 spark-submit --kill 选项来关闭，但前提是 Spark Streaming 程序需要设置 spark.streaming.stopGracefullyOnShutdown=true
• 使用一个标记文件（marker file）, 比如是存储在 HDFS 上的一个空文件，程序定期地监控这个文件，如果此文件不存在，就调用 streamingContext.stop(stopSparkContext = true, stopGracefully = true)，作者更加推荐这个方式