這篇文章記錄我使用 Spark Streaming 進行 ETL 處理的總結，主要包含如何編程，以及遇到的問題。
環境
我在公司使用的環境如下：
- Spark: 2.2.0
- Kakfa: 0.10.1
這兩個版本算是比較新的。
業務
從 Kafka 中讀取數據，用 SQL 處理，寫入 Kafka 中。 程序主要分為 3大塊：
- 從 Kafka 中讀取數據。
- SQL ETL。
- 寫入 Kafka。
編程
從 Kafka 中讀取數據
spark-streaming-kafka-0-10_2.11
最開始使用spark-streaming-kafka-0-10_2.11。雖然這個包是實驗階段，但是考慮到用起來比較方便，就使用了這個包。整個代碼的框架和官方文檔的一樣。
stream.foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// some time later, after outputs have completed
stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
編程很快，但是后面遇到了很多問題:
-
異常錯誤：WARN TaskSetManager: Lost task 9.0 in stage 1683.0 (TID 9460, 10.62.34.25, executor 9): java.lang.AssertionError: assertion failed: Failed to get records for spark-executor-2017-10-20-1100-streaming-test 1231231 1 13733588428 after polling for 1000。
這個錯誤是 DirectKafkaStream 在 poll 數據的時候，發現沒有數據返回, 代碼如下：
```scala // 從 buffer 獲取數據，如果buffer 中沒有數據，就 poll 數據。 if (!buffer.hasNext()) { poll(timeout) } assert(buffer.hasNext(), s"Failed to get records for $groupId $topic $partition $offset after polling for $timeout") var record = buffer.next() ... ```上面的代碼的意思是從 kafka 中 poll 數據，如果 timeout 長時間后還沒有得到數據，就報錯。 而實際我們的 Kafka 數據每秒鍾有幾千條。 並且 timeout 默認是 1秒，不可能拿不到數據。最后發現
spark-streaming-kafka-0-10_2.11這個包對應的kafka-clients是 0.10.0.1。而這個版本的kafka-clients是有 BUG的，於是將kafka-clients的版本升級到 0.10.2.1。問題解決了。 -
測試的時候，發現在停止掉程序后，在重開程序，重復消費一部分數據。 那么這個問題就是，程序停止的時候沒有正確的提交當前消費的 offset。
我們的程序是通過stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)來提交每個 RDD 的 offset 的。而這段代碼的背后是將 offsetRanges 保存到了一個隊列中。 等到下次從 kafka 中獲取下一個 batch 的數據后（通過 compute 函數），順便將隊列中的 offset 提交到 KafkaCluster 中。
代碼如下：// 保存到 queue 中 def commitAsync(offsetRanges: Array[OffsetRange], callback: OffsetCommitCallback): Unit = { commitCallback.set(callback) commitQueue.addAll(ju.Arrays.asList(offsetRanges: _*)) } // 提交 offset , 將 queue 中的 offset 保存到 map 中，並提交 protected def commitAll(): Unit = { val m = new ju.HashMap[TopicPartition, OffsetAndMetadata]() var osr = commitQueue.poll() while (null != osr) { val tp = osr.topicPartition val x = m.get(tp) val offset = if (null == x) { osr.untilOffset } else { Math.max(x.offset, osr.untilOffset) } m.put(tp, new OffsetAndMetadata(offset)) osr = commitQueue.poll() } if (!m.isEmpty) { consumer.commitAsync(m, commitCallback.get) } } // 每次從 kafka 中獲取數據, 順便提交 上一次的 offset override def compute(validTime: Time): Option[KafkaRDD[K, V]] = { // 獲取當前的 offset, 如果程序保存了offset就用程序的，如果沒有，就從kafka中讀取。 // 當程序重啟后，就會從kafka中讀取。 val untilOffsets = clamp(latestOffsets()) val offsetRanges = untilOffsets.map { case (tp, uo) => val fo = currentOffsets(tp) OffsetRange(tp.topic, tp.partition, fo, uo) } ... // 獲取到了數據，並保存在 rdd 中 val rdd = new KafkaRDD[K, V](context.sparkContext, executorKafkaParams, offsetRanges.toArray, getPreferredHosts, useConsumerCache) .... // 更新 offset currentOffsets = untilOffsets // 重點：提交 queue 中的offset commitAll() Some(rdd) }看完這個邏輯，傻眼了。這樣子程序結束，處理完最后一個 batch， 它的 offset 是沒有辦法提交到 cluster 的，結果就是重復消費。如果要自己寫提交 offset 的代碼，那和老版本的就沒有區別了。
考慮了半天，最終還是用老的包來實現了。
spark-streaming-kafka-0-8
使用老的包，我們的邏輯如下：
代碼實現如下：
...
各種參數初始化
val kafkaCluster = new KafkaCluster(kafkaClusterParams)
val topicAndPartitionSet = kafkaCluster.getPartitions(consumerTopics.toSet).right.get
var consumerOffsetsLong = new mutable.HashMap[TopicAndPartition, Long]()
if (kafkaCluster.getConsumerOffsets(kafkaClusterParams.get("group.id").toString, topicAndPartitionSet).isLeft) {
val latestOffset = kafkaCluster.getLatestLeaderOffsets(topicAndPartitionSet)
topicAndPartitionSet.foreach(tp => {
consumerOffsetsLong.put(tp, latestOffset.right.get(tp).offset)
})
} else {
val consumerOffsetsTemp = kafkaCluster.getConsumerOffsets(kafkaClusterParams.get("group.id").toString, topicAndPartitionSet)
topicAndPartitionSet.foreach(tp => {
consumerOffsetsLong.put(tp, consumerOffsetsTemp.right.get(tp))
})
}
val kafkaClusterParamsBroadcast = ssc.sparkContext.broadcast(kafkaClusterParams)
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
ssc, kafkaClusterParams, consumerOffsetsLong.toMap, (mmd: MessageAndMetadata[String, String]) => mmd.message() )
stream.foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// 處理業務邏輯
val m = new mutable.HashMap[TopicAndPartition, Long]()
if (null != offsetRanges) {
offsetRanges.foreach(
osr => {
val tp = osr.topicAndPartition
m.put(tp, osr.untilOffset)
}
)
}
kafkaCluster.setConsumerOffsets(kafkaClusterParamsBroadcast.value.get("group.id").toString, m.toMap)
}
這樣子來處理數據，一切正常。
SQL ETL
SQL ETL 就是使用 Spark SQL 進行處理。如果要對多個同一個 batch 進行多次處理，最好是 將 bacth cache 起來。
將數據寫入 Kafak 中
這個就是從網上找的了：
import java.util.concurrent.Future
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerRecord, RecordMetadata }
class KafkaSink[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
/* This is the key idea that allows us to work around running into
NotSerializableExceptions. */
lazy val producer = createProducer()
def send(topic: String, key: K, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, key, value))
def send(topic: String, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, value))
def close(): Unit = {
producer.close()
}
}
object KafkaSink {
import scala.collection.JavaConversions._
def apply[K, V](config: Map[String, AnyRef]): KafkaSink[K, V] = {
val createProducerFunc = () => {
val producer = new KafkaProducer[K, V](config)
producer
}
new KafkaSink(createProducerFunc)
}
def apply[K, V](config: java.util.Properties): KafkaSink[K, V] = apply(config.toMap)
}
使用方式:
// 廣播KafkaSink
val kafkaSinkBroadcast: Broadcast[KafkaSink[String, String]] = {
ssc.sparkContext.broadcast(KafkaSink[String, String](kafkaSinkParams))
}
val kafkaProducerTopicBroadcast = ssc.sparkContext.broadcast(producerTopic)
stream.foreachRDD {
....
kafkaSinkBroadcast.value.send(kafkaProducerTopicBroadcast.value, str)
}
整體上的代碼就是這么多。
配置
除了代碼，Spark Streaming 還是需要某些配置的，具體如下：
- "spark.executor.cores":"2"。默認的 Yarn 模式下，core 的個數是1個。當 executor 的壓力過大的時候，經常會出現 connect reset by peer 和 心跳超時，所以要看情況增加 core 的個數。
- "spark.driver.extraJavaOptions":"-Dlog4j.configuration=file:log4j.properties" 。Spark 默認的日志級別就是 INFO, 通常會打印出很多的信息，日志一晚上就上G了，所以最好自定義自己的配置文件。
- "spark.executor.extraJavaOptions":"-XX:+UseG1GC -XX:+PrintGCDetails -XX:+PrintHeapAtGC -XX:+PrintGCTimeStamps" 。 使用 G1 的垃圾回收方式，並打印出具體的信息，方便在 GC 時間過長的時候進行調優。
- "spark.streaming.stopGracefullyOnShutdown":"true"。讓 Streaming 程序在收到 Terminate 信號后，處理完最后一個 batch 再退出。通常停止程序的時候，運行兩次
kill -15 driver_pid就可以停止掉程序。
"spark.streaming.backpressure.enabled":"true",
"spark.streaming.backpressure.initialRate":"1000000",
"spark.streaming.kafka.maxRatePerPartition":"20000",
這三個參數用來限制消費 kafka 的速度。避免一次消費太多的數據，將程序搞垮掉。