title: How to write the input CSV from Kafka topic to a postgres table using Pyspark?
body: <p>I am new to spark and spark streaming and I started out with this task where I create a kafka topic and use a python producer script to input some data into the topic. The py script reads a csv file and writes to the topic.</p>
  <pre><code>from kafka import KafkaProducer
  import time
  import json
  from json import dumps
  import csv
  
  
  producer = KafkaProducer(bootstrap_servers = ['localhost:9092'],value_serializer=lambda x: dumps(x).encode('utf-8'))
  with open('/home/yashdeep/test_data.csv') as file:
      reader = csv.DictReader(file, delimiter=&quot;,&quot;)
      for row in reader:
          producer.send(topic='testdata', value=row)
          time.sleep(1)
  </code></pre>
  <p>When I used the kafka consumer shell script to check if the topic has any messages I could see that all the messages are in a JSON/ dict format. I then used pyspark to read this data from the kafka topic to a dataframe. I printed the results using console sink. Now my problem is I don't know how to proceed further. I want write this streamed data to a postgres db table. I tried to see through the documentation but I am having trouble understanding to do so. I have downloaded the postgresql jar and added to the jars folder in the spark directory. So far the code I was able to write is this.</p>
  <pre><code>from pyspark.sql import SparkSession
  from pyspark.sql.functions import *
  from pyspark.sql.types import *
  
  PSQL_HOSTNAME = &quot;localhost&quot;
  PSQL_PORTNUMBER = 5432
  PSQL_DBNAME = &quot;mydb&quot;
  PSQL_TABLENAME = 't1'
  PSQL_USERNAME = &quot;myuser&quot;
  PSQL_PASSWORD = &quot;******&quot;
  PSQL_JDBC_URL = &quot;jdbc:postgresql://&quot; + PSQL_HOSTNAME + &quot;:&quot; + PSQL_PORTNUMBER + &quot;/&quot; + 
  PSQL_DBNAME
  
  spark = SparkSession.builder\
      .appName(&quot;Sample test&quot;)\
      .master(&quot;local[*]&quot;)\
      .getOrCreate()
  
  spark.sparkContext.setLogLevel(&quot;ERROR&quot;)
  
  df = spark.readStream\
      .format(&quot;kafka&quot;)\
      .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:9092&quot;)\
      .option(&quot;subscribe&quot;, &quot;testdata&quot;)\
      .option(&quot;startingoffsets&quot;, &quot;earliest&quot;)\
      .load()
  
  df1 = df.selectExpr(&quot;CAST(value AS STRING)&quot;, &quot;timestamp&quot;)
  write_stream = df1 \
      .writeStream \
      .trigger(processingTime='5 seconds') \
      .outputMode(&quot;update&quot;) \
      .option(&quot;truncate&quot;, &quot;false&quot;)\
      .format(&quot;console&quot;) \
      .start()
      
  write_stream.awaitTermination()
  </code></pre>
  <p>Please help me. If there is any other post that has the answer or any useful links to do this please share it. Also if there is a way to connect to postgres using pyscopg2 and write data let me know. Thank You in advance</p>