我正在使用Spark 2.1.0和Kafka 0.9.0。
我试图将批处理Spark作业的输出推送到kafka。该作业应每小时运行一次,但不是作为流处理。
在网上寻找答案时,我只能找到与Spark流集成相关的kafka,没有关于批处理作业集成的内容。
有人知道这种事情是否可行吗?
谢谢
更新:
如用户8371915所述,我尝试按照 Writing the output of Batch Queries to Kafka中的步骤进行操作。
我使用了一个spark shell:
但我收到了错误提示:
我试图将批处理Spark作业的输出推送到kafka。该作业应每小时运行一次,但不是作为流处理。
在网上寻找答案时,我只能找到与Spark流集成相关的kafka,没有关于批处理作业集成的内容。
有人知道这种事情是否可行吗?
谢谢
更新:
如用户8371915所述,我尝试按照 Writing the output of Batch Queries to Kafka中的步骤进行操作。
我使用了一个spark shell:
spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
这是我尝试的简单代码:
val df = Seq(("Rey", "23"), ("John", "44")).toDF("key", "value")
val newdf = df.select(to_json(struct(df.columns.map(column):_*)).alias("value"))
newdf.write.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("topic", "alerts").save()
但我收到了错误提示:
java.lang.RuntimeException: org.apache.spark.sql.kafka010.KafkaSourceProvider does not allow create table as select.
at scala.sys.package$.error(package.scala:27)
at org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala:497)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:215)
... 50 elided
有任何想法这是与什么相关的吗?
谢谢