Structured Streaming从Kafka 0.8中读取数据的问题

众所周知，Structured Streaming默认支持Kafka 0.10，没有提供针对Kafka 0.8的Connector，但这对高手来说不是事儿，于是有个Hortonworks的邵大牛（前段时间刚荣升Spark Committer）给出了一个开源的第三方解决方案：

https://github.com/jerryshao/spark-kafka-0-8-sql

不过下载下来后，编译不通过：



看来有时候，牛人办事也不一定靠谱，当然从github中的issue记录看，邵大牛认为是spark方面做了改动：

https://github.com/jerryshao/spark-kafka-0-8-sql/issues/7

不过想基于Structured Streaming使用Kafka 0.8的人不在少数，另有祝威廉把这个问题给修正了：

https://github.com/allwefantasy/streamingpro/tree/master/streamingpro-spark-2.0/src/main/java/com/hortonworks/spark/sql/kafka08

使用方式：

val kafkaDf = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "10.20.0.1:9093")
.option("startingoffset", "latest")
.option("subscribe", engineConfig.sourceTopic)
.load()

由于Kafka 0.10才提供了时间戳功能，所以尽管经过上面的折腾，Structured Streaming可以和Kafka 0.8进行Connect，但是从Kafka 0.8中读取的记录，默认是不带TimeStamp的！

从代码中，我们可以看到，只有如下5个数据类型：

StructField("key", BinaryType),
StructField("value", BinaryType),
StructField("topic", StringType),
StructField("partition", IntegerType),
StructField("offset", LongType)

如果要使用Structured Streaming的基于Event Time功能，以及Watermark特性，由于没法直接从Kafka 0.8的数据中提取到TimeStamp，需要通过转换进行内部记录时间戳的提取。

再次感谢邵大牛及祝大牛的无私奉献！