Structured Streaming从Kafka 0.8中读取数据的问题
2024-10-18 22:34:30
众所周知,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,需要通过转换进行内部记录时间戳的提取。
再次感谢邵大牛及祝大牛的无私奉献!
最新文章
- ProgressBar---进度条
- 记录一下折腾webp 的过程
- check fasta format
- ORACLE迁移votedisk,spfile以及OCRfile的方法
- Spring AOP 由浅入深
- 关于http请求
- PE渲染引擎 二
- 多线程基础(二)pthread的了解
- 修剪花卉(codevs 1794)
- 04_HttpClient发送Https请求
- ios常见加密解密方法
- NLP 苏图南 打破自我设限 突破自我—在线播放—优酷网,视频高清在线观看
- MVC3+EF4.1学习系列(二)-------基础的增删改查和持久对象的生命周期变化
- Java集合总结系列2:Collection接口
- Java学习笔记16---抽象类与接口的浅显理解
- linux链接
- Text Mining and Analytics WEEK1
- Oracle索引之Btree索引
- Java面试(一) -- 基础部分(1)
- extjs_10_自己定义combotree组件