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

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

• 从Kafka读取数据

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)]

  }

• Kafka中的基础信息字段和类型如下：

最后

以上就是爱撒娇蛋挞最近收集整理的关于深入浅出Spark实时处理（三）--- Structured Streaming集成Kafka的全部内容，更多相关深入浅出Spark实时处理（三）---内容请搜索靠谱客的其他文章。