在 Spark Streaming (Spark 2.0) 中使用 Kafka

Question

我发现有两种方法可以在 Spark Streaming (Spark 2.0) 中使用 Kafka 主题：

1) 使用KafkaUtils.createDirectStream每k秒获取一次DStream，请参考this document

2) 使用 kafka: sqlContext.read.format(“json”).stream(“kafka://KAFKA_HOST”) 为 Spark 2.0 的新功能创建无限数据帧：结构化流，相关文档 is here

方法 1) 有效，但方法 2) 无效，出现以下错误

Exception in thread "main" java.lang.NoSuchMethodError: org.apache.spark.sql.DataFrameReader.stream(Ljava/lang/String;)Lorg/apache/spark/sql/Dataset;
...
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:729)
at org.apache.spark.deploy.SparkSubmit$.doRunMain(SparkSubmit.scala:185)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我的问题是：
“kafka://KAFKA_HOST” 指的是什么？
我应该如何解决这个问题？

提前致谢！

Answer 1

Spark 2.0 尚不支持 Kafka 作为无限源 DataFrames/Sets。计划在 2.1 中添加支持

编辑：(6.12.2016)

Kafka 0.10 现在是 expiramentaly supported in Spark 2.0.2:

val ds1 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()

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

在 Spark Streaming (Spark 2.0) 中使用 Kafka

Consume Kafka in Spark Streaming (Spark 2.0)

apache-kafka

apache-spark

spark-streaming

apache-spark-sql

spark-structured-streaming

编辑：(6.12.2016)