Question: I am using Spark 2.1.0 and Kafka 0.9.0.

I am trying to push the output of a batch spark job to kafka. The job is supposed to run every hour but not as streaming.

While looking for an answer on the net I could only find kafka integration with Spark streaming and nothing about the integration with the batch job.

Does anyone know if such thing is feasible ?

Thanks

UPDATE :

As mentioned by user8371915, I tried to follow what was done in [Writing the output of Batch Queries to Kafka](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#writing-the-output-of-batch-queries-to-kafka).

I used a spark shell :

```
spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0

```

Here is the simple code that I tried :

```
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()

```

But I get the error :

```
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

```

Have any idea what is this related to ?

Thanks

Answer: 
**tl;dr** You use outdated Spark version. Writes are enabled in 2.2 and later.

Out-of-the-box you can use Kafka SQL connector (the same as used with Structured Streaming). Include

* `spark-sql-kafka` in your dependencies.
* Convert data to `DataFrame` containing at least `value` column of type `StringType` or `BinaryType`.
* Write data to Kafka:

```
df   
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", server)
  .save()

```

Follow [Structured Streaming docs](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html) for details (starting with [Writing the output of Batch Queries to Kafka](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#writing-the-output-of-batch-queries-to-kafka)).
If you have a dataframe and you want to write it to a kafka topic, you need to convert columns first to a "value" column that contains data in a json format. In scala it is

```scala
import org.apache.spark.sql.functions._

val kafkaServer: String = "localhost:9092"
val topicSampleName: String = "kafkatopic"

df.select(to_json(struct("*")).as("value"))
  .selectExpr("CAST(value AS STRING)")
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaServer)
  .option("topic", topicSampleName)
  .save()

```