深入浅出Spark实时处理（三）--- Structured Streaming集成Kafka

98 阅读 0 评论 65 点赞

我是靠谱客的博主爱撒娇蛋挞，这篇文章主要介绍深入浅出Spark实时处理（三）--- Structured Streaming集成Kafka，现在分享给大家，希望可以做个参考。

Kafka 0.10的 Structured Streaming 集成，可从Kafka读取数据或向Kafka写入数据。

从Kafka读取数据

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def run(): Unit ={

      val df = spark.readStream

        .format("kafka")

        .option("kafka.bootstrap.servers", "host1:port1,host2:port2")

        .option("subscribe", "topic1")

        .load()



      df.selectExpr("CAST(key as STRING)", "CAST(value AS STRING)").as[(String, String)]



      val df = spark

        .readStream

        .format("kafka")

        .option("kafka.bootstap.servers", "host1:port1,host2:port2")

        .option("subscribe", "topic1,topic2")

        .load()

    

      df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").as[(String, String)]

  }