Spark Streaming消息的传输与保证及编程示例

Spark streaming的文档其实已经比较详细的介绍了， 我简单的复述一次。   只要涉及到消息，就包含2个阶段， 获取数据以及存储处理后的数据。Spark高级API采用WAL + ZK的方式保证at least once（不能保证exactly once),  获取数据之后spark会先把数据写入WAL，并把拿到的offset存储到ZK里面。 处理完数据之后然后返回确认机制，表示消息已经发送成功。 如果是work失败了， spark只需要通过driver重新获取数据即可，如果是driver失败，很显然所有executor丢失，数据也同样丢失，这个时候如果使用了wal，那么只需要重启任务，从WAL获取上次的信息再处理，很显然这就是at
least once。 

那么at most once什么时候出现呢？ 那就是你不使用WAL， 因为不使用WAL， 即使有消息处理不成功，driver一旦失败，数据丢失，状态信息全无，spark也表示无能为力了，因为没有信息可以恢复这些数据。

exactly once使用高级API是不能实现的，只能使用低级API，也就是createdirectstream，低级API有几个功能很有用，比如自己指定offset消费，从我的测试来看，高级API初次获取数据只能从largest offset拿数据，不能够从latest offset开始处理，这个时候就要用低级API来处理了。低级API的offset默认是不保存在ZK的，为了避免性能问题也不使用WAL。

上面这些理论实际和storm有类似的地方，不同之处在于2者在处理上面3种模式的方法不同，storm才用的ACK + POP机制，而spark才用的是WAL。 另外storm是实时获取数据，而spark是才用频率的模式获取数据，所以2者在这个地方还有比较细微的差别。

下面我简单的写了一个程序，从KAFKA获取数据，然后不做任何处理，直接存储到HBASE。

package com.isesol.spark

import kafka.utils.ZkUtils
import kafka.serializer.Decoder
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import java.util.HashMap
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerConfig, ProducerRecord }
import org.apache.spark.streaming.kafka._
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.spark._
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.filter._
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
import org.apache.hadoop.hbase.client.HTable
import org.apache.hadoop.hbase.HBaseConfiguration

object high_streaming {
def main(args: Array[String]) {
val conf = new SparkConf().setMaster("yarn-cluster").setAppName("this is the first spark streaming program!")
val ssc = new StreamingContext(conf, Seconds(5))
ssc.checkpoint("hdfs://nameservice1/tmp/high_streaming1")
val zk = "datanode01.isesol.com,datanode02.isesol.com,datanode03.isesol.com,datanode04.isesol.com,cmserver.isesol.com"
val group = "stream_2001"
val topics = "2001"
val numThreads = 2
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zk, group, topicMap).map(_._2)
// lines.foreachRDD(rdd => rdd.foreach { x => println(x) })
val tablename = "test"
lines.foreachRDD { rdd =>
rdd.foreachPartition { x =>
val hbaseconf = HBaseConfiguration.create()
hbaseconf.set("hbase.zookeeper.quorum", "datanode01.isesol.com,datanode02.isesol.com,datanode03.isesol.com,datanode04.isesol.com,cmserver.isesol.com")
hbaseconf.set("hbase.zookeeper.property.clientPort", "2181")
val myTable = new HTable(hbaseconf, TableName.valueOf(tablename))
//myTable.setAutoFlush(false)
myTable.setWriteBufferSize(3 * 1024 * 1024)
x.foreach { y =>
{
println(y)
val p = new Put(Bytes.toBytes(System.currentTimeMillis().toString()))
p.add(Bytes.toBytes("cf"), Bytes.toBytes("message"), Bytes.toBytes(y.toString()))
myTable.put(p)
}
}
myTable.close()
}
}

ssc.start()
ssc.awaitTermination()

}
}

通过命令行提交, 提交JOB的时候记得把spark-streaming-kafka的包带进去，否者会提示找不到类.：

spark-submit --class com.isesol.spark.high_streaming  --master yarn --deploy-mode cluster --jars spark-streaming-kafka_2.10-1.6.0-cdh5.9.0.jar --driver-memory 1g --executor
9f7d
-memory 1G    --conf spark.streaming.receiver.writeAheadLog.enable=true  --num-executors 5  high_streaming.jar

通过SPARK UI可以监控到启动的程序:





这里主要关注的点是处理的性能，也就是processing time（处理时间）， 以及schedule delay (上一个batch如果没有处理完，下一个batch需要等待的时间）， 从这2个时间就能看出spark处理数据是否有延迟，如果schedule delay如果时间超过秒级，我们就认为有延迟，如果毫秒级，我倒觉得这个是正常。

这篇文章讲述的是的消息和传输机制，以及一个简单的示例程序， 用来感受一下spark streaming到底是一个啥玩意，实际上来说比较简单，包括storm程序的开发也是如此，这个程序唯一和生产环境JOB不同的是中间的处理逻辑，因为我没有处理，只是直接写入，生产环境你会运用各种不同的SPARK方法来处理数据，再存储，仅仅如此而已。