Driver Out of Memory for a Spark Structured Streaming Job

I am trying to run a file-based Structured Streaming job with S3 as a source. These files are in JSON format. The code reads the newly added files to the S3 folder, parse the JSON attributes and writes the data back to S3 in parquet format. I am running the Spark job on a AWS EMR cluster (version - 5.29.0)

Streaming Code

def writeToOutput(inputDF, batchId):
    spark.sql("drop table if exists global_temp.source_df")

    inputDF.cache()

    inputDF.createGlobalTempView("source_df")

    df1 = spark.sql(df1_sql)
    df2 = spark.sql(df2_sql)
    df3 = spark.sql(df3_sql)    

    df1.repartition(1) \
    .write \
    .partitionBy("col1", "col2") \
    .format("parquet") \
    .mode('append') \
    .save(output_path + 'df1/')

    df2.repartition(1) \
    .write \
    .partitionBy("col1", "col2") \
    .format("parquet") \
    .mode('append') \
    .save(output_path + 'df2/')

    df3.repartition(1) \
    .write \
    .partitionBy("col1", "col2") \
    .format("parquet") \
    .mode('append') \
    .save(output_path + 'df3/')

    inputDF.unpersist()


inputDF = spark \
          .readStream \
          .schema(jsonSchema) \
          .option("latestFirst", "false") \
          .option("badRecordsPath", bad_records_path) \
          .option("maxFilesPerTrigger", "2000") \
          .json(input_path).withColumn('file_path', input_file_name())


query = inputDF.writeStream \
        .foreachBatch(writeToOutput) \
        .queryName("Stream") \
        .option("checkpointLocation", checkpoint_path) \
        .trigger(processingTime='180 seconds') \
        .start()

query.awaitTermination()

My spark-submit configs are: spark-submit --master yarn --deploy-mode cluster --executor-memory 5G --executor-cores 4 --driver-memory 15G --num-executors 40 --conf spark.dynamicAllocation.enabled=false --conf yarn.resourcemanager.am.max-attempts=4 --conf spark.yarn.am.attemptFailuresValidityInterval=1h --conf spark.executor.memoryOverhead=1024 --conf spark.driver.memoryOverhead=512 --conf spark.yarn.max.executor.failures=300 --conf spark.yarn.executor.failuresValidityInterval=1h --conf spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2 --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.streaming.stopGracefullyOnShutdown=true --conf spark.task.maxFailures=8 --conf spark.driver.extraJavaOptions=-Dlog4j.configuration=log4j-yarn.properties --conf spark.executor.extraJavaOptions=-Dlog4j.configuration=log4j-yarn.properties

My jobs run for 1-2 hours and fail with the error Container is running beyond physical memory limits. Current usage: 15.6 GB of 15.5 GB physical memory used; 19.1 GB of 77.5 GB virtual memory used. Killing container.

I cannot understand why the driver needs so much of memory? Can someone please help