小编Azz*_*zzy的帖子

我使用的是Spark 2.1.0和Kafka 0.9.0.

我试图将批量火花作业的输出推送到kafka.这项工作应该每小时运行一次,但不是作为流媒体运行.

在网上寻找答案时,我只能找到kafka与Spark流的集成,而不是与批处理作业的集成.

有谁知道这样的事情是否可行？

谢谢

更新:

正如user8371915所提到的,我试图遵循将批量查询的输出写入Kafka所做的工作.

我用了火花壳:

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

知道这与此有关吗？

谢谢