title: How to write from Spark strucutred streaming to RDMS? What is inside foreach row?
body: <p>It is <a href="https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreach" rel="nofollow noreferrer">written in documentation</a>, that one can write rows to database with <code>foreach</code> against <code>writeStream</code></p>
  <pre><code>def process_row(row):
      # Write row to storage
      pass
  
  query = streamingDF.writeStream.foreach(process_row).start()
  </code></pre>
  <p>What is not written, is what is inside <code>row</code>? How can I process it without this knowledge?</p>
  <hr />
  <p>Currently I wrote</p>
  <pre><code>df = df... # something
  
  def write_row(row):
  
      mydb = mysql.connector.connect(
          host=config[&quot;jdbc&quot;][&quot;host&quot;],
          port=config[&quot;jdbc&quot;][&quot;port&quot;],
          user=config[&quot;jdbc&quot;][&quot;user&quot;],
          password=config[&quot;jdbc&quot;][&quot;password&quot;],
          database=config[&quot;jdbc&quot;][&quot;database&quot;],
      )
  
      dbtable = ...
  
      mycursor = mydb.cursor()
  
      sql = f&quot;INSERT INTO {dbtable} (user_id, carName) VALUES (%s, %s)&quot;
      val = (352, &quot;Sample&quot;)
      mycursor.execute(sql, val)
  
      mydb.commit()
  
  
  query = df \
      .writeStream \
      .outputMode('update') \
      .foreach(write_row) \
      .start()
  
  query.awaitTermination()
  </code></pre>
  <p>Unfortunately, it doesn't write anything. Function <code>write_row</code> itself works and writes sample row. When inside <code>foreach</code>, it doesn't.</p>
  <p>If error occurs in this function, how would I know it?</p>
  <hr />
  <p>I rewrote according to @AlexOtt.</p>
  <p>If I run this</p>
  <pre><code>def foreach_batch_function(df1, epoch_id):
      df1.write.format(&quot;jdbc&quot;) \
          .option(&quot;url&quot;, &quot;jdbc:mariadb://IPADDRESS/database&quot;) \
          .option(&quot;dbtable&quot;, &quot;pysparktest&quot;) \
          .option(&quot;user&quot;, config[&quot;jdbc&quot;][&quot;user&quot;]) \
          .option(&quot;password&quot;, config[&quot;jdbc&quot;][&quot;password&quot;]) \
          .option(&quot;driver&quot;, &quot;org.mariadb.jdbc.Driver&quot;) \
          .save()
  </code></pre>
  <p>or with option</p>
  <pre><code>.option(&quot;createTableColumnTypes&quot;, &quot;cnt int, minBestLapTime double, trackName varchar(64), trackVersion varchar(64), carClass varchar(64), carName varchar(64), carVersion varchar(64)&quot;) \
  </code></pre>
  <p>while table does not exist, it fails with SQL syntax error. And if table already exists, it fails saying &quot;table exists&quot;.</p>
  <p>Is it possible to populate existing table?</p>