title: How do I write to Kafka using pyspark?
body: <p>I am trying to write to Kafka using PySpark.<br>
  I got stuck on stage zero:</p>
  
  <pre><code>[Stage 0:&gt;                                                          (0 + 8) / 9]
  </code></pre>
  
  <p>Then I get a timeout error:</p>
  
  <p><code>org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.</code></p>
  
  <p>Code is:</p>
  
  <pre><code>import os
  os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages
   org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 pyspark-shell'
  
  from pyspark.sql.functions import *
  from pyspark.sql import SparkSession
  from pyspark.sql.types import *
  
  def main():
      spark = SparkSession.builder.master("local").appName("Spark CSV Reader")
       .getOrCreate();
  
      dirpath =  os.path.abspath(sys.argv[1])
      os.chdir(dirpath)
  
      mySchema = StructType([
       StructField("id", IntegerType()),StructField("name", StringType()),\
       StructField("year", IntegerType()),StructField("rating", DoubleType()),\
       StructField("duration", IntegerType())   ])
      streamingDataFrame = spark.readStream.schema(mySchema)
       .csv('file://' + dirpath + "/" )
  
      streamingDataFrame.selectExpr("CAST(id AS STRING) AS key",
       "to_json(struct(*)) AS value").\
        writeStream.format("kafka").option("topic", "topicName")\
        .option("kafka.bootstrap.servers", "localhost:9092")\
        .option("checkpointLocation", "./chkpt").start()
  </code></pre>
  
  <p>I am running HDP 2.6.</p>