How to write the input CSV from Kafka topic to a postgres table using Pyspark?

Question

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.

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=",")
    for row in reader:
        producer.send(topic='testdata', value=row)
        time.sleep(1)

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.

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

PSQL_HOSTNAME = "localhost"
PSQL_PORTNUMBER = 5432
PSQL_DBNAME = "mydb"
PSQL_TABLENAME = 't1'
PSQL_USERNAME = "myuser"
PSQL_PASSWORD = "******"
PSQL_JDBC_URL = "jdbc:postgresql://" + PSQL_HOSTNAME + ":" + PSQL_PORTNUMBER + "/" + 
PSQL_DBNAME

spark = SparkSession.builder\
    .appName("Sample test")\
    .master("local[*]")\
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

df = spark.readStream\
    .format("kafka")\
    .option("kafka.bootstrap.servers", "localhost:9092")\
    .option("subscribe", "testdata")\
    .option("startingoffsets", "earliest")\
    .load()

df1 = df.selectExpr("CAST(value AS STRING)", "timestamp")
write_stream = df1 \
    .writeStream \
    .trigger(processingTime='5 seconds') \
    .outputMode("update") \
    .option("truncate", "false")\
    .format("console") \
    .start()
    
write_stream.awaitTermination()

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

How to write the input CSV from Kafka topic to a postgres table using Pyspark?

0 Answers