title: Issue in writing records in into MYSQL from Spark Structured Streaming Dataframe
body: <p>I am using  below code to write spark Streaming dataframe into MQSQL DB .Below is the kafka topic JSON data format and MYSQL table schema.Column name and types are same to same.
  But I am unable to see records written in MYSQL table. Table is empty with zero records.Please suggest.</p>
  <p>Kafka Topic Data Fomat<br />
  ssingh@RENLTP2N073:/mnt/d/confluent-6.0.0/bin$ ./kafka-console-consumer --topic sarvtopic --from-beginning --bootstrap-server localhost:9092
  {&quot;id&quot;:1,&quot;firstname&quot;:&quot;James &quot;,&quot;middlename&quot;:&quot;&quot;,&quot;lastname&quot;:&quot;Smith&quot;,&quot;dob_year&quot;:2018,&quot;dob_month&quot;:1,&quot;gender&quot;:&quot;M&quot;,&quot;salary&quot;:3000}
  {&quot;id&quot;:2,&quot;firstname&quot;:&quot;Michael &quot;,&quot;middlename&quot;:&quot;Rose&quot;,&quot;lastname&quot;:&quot;&quot;,&quot;dob_year&quot;:2010,&quot;dob_month&quot;:3,&quot;gender&quot;:&quot;M&quot;,&quot;salary&quot;:4000}</p>
  <pre><code>import pyspark
  
  from pyspark.sql import SparkSession
  
  spark = SparkSession \
      .builder \
      .appName(&quot;SSKafka&quot;) \
      .getOrCreate()
      dsraw = spark \
    .readStream \
    .format(&quot;kafka&quot;) \
    .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:9092&quot;) \
    .option(&quot;subscribe&quot;, &quot;sarvtopic&quot;) \
    .option(&quot;startingOffsets&quot;, &quot;earliest&quot;) \
    .load()
  
  ds = dsraw.selectExpr(&quot;CAST(value AS STRING)&quot;)
  dsraw.printSchema()
  
  from pyspark.sql.types import StructField, StructType, StringType,LongType
  from pyspark.sql.functions import *
  
  custom_schema = StructType([
      StructField(&quot;id&quot;, LongType(), True),
      StructField(&quot;firstname&quot;, StringType(), True),
      StructField(&quot;middlename&quot;, StringType(), True),
      StructField(&quot;lastname&quot;, StringType(), True),
      StructField(&quot;dob_year&quot;, StringType(), True),
      StructField(&quot;dob_month&quot;, LongType(), True),
      StructField(&quot;gender&quot;, StringType(), True),
      StructField(&quot;salary&quot;, LongType(), True),
  ])
        
  Person_details_df2 = ds\
          .select(from_json(col(&quot;value&quot;), custom_schema).alias(&quot;Person_details&quot;))
          
  Person_details_df3 = Person_details_df2.select(&quot;Person_details.*&quot;)
  
  
  from pyspark.sql import DataFrameWriter
  
  def foreach_batch_function(df, epoch_id):
      Person_details_df3.write.jdbc(url='jdbc:mysql://172.16.23.27:30038/securedb', driver='com.mysql.jdbc.Driver', dbtable=&quot;sparkkafka&quot;,  user='root',password='root$1234')
      pass
  
  query = Person_details_df3.writeStream.trigger(processingTime='20 seconds').outputMode(&quot;append&quot;).foreachBatch(foreach_batch_function).start()
  
  query
  Out[14]: &lt;pyspark.sql.streaming.StreamingQuery at 0x1fb25503b08&gt;
  
  MYSQL table Schema:
  
  create table sparkkafka(
     id int,
     firstname VARCHAR(40) NOT NULL,
     middlename VARCHAR(40) NOT NULL,
     lastname VARCHAR(40) NOT NULL,
     dob_year int(40) NOT NULL,
     dob_month int(40) NOT NULL,
     gender VARCHAR(40) NOT NULL,
     salary int(40) NOT NULL,   
     PRIMARY KEY (id)
  );
  </code></pre>