title: How to use foreach or foreachBatch in PySpark to write to database?
body: <p>I want to do Spark Structured Streaming (Spark 2.4.x) from a Kafka source to a MariaDB with Python (PySpark).</p>
  
  <p>I want to use the streamed Spark dataframe and not the static nor Pandas dataframe.</p>
  
  <p>It seems that one has to use <code>foreach</code> or <code>foreachBatch</code> since there are no possible database sinks for streamed dataframes according to <a href="https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks" rel="noreferrer">https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks</a>.</p>
  
  <p>Here is my try:</p>
  
  <pre><code>from pyspark.sql import SparkSession
  import pyspark.sql.functions as F
  from pyspark.sql.types import StructField, StructType, StringType, DoubleType, TimestampType
  from pyspark.sql import DataFrameWriter
  # configuration of target db
  db_target_url = "jdbc:mysql://localhost/database"
  db_target_properties = {"user":"writer", "password":"1234"}
  # schema
  schema_simple = StructType([StructField("Signal", StringType()),StructField("Value", DoubleType())])
  
  # create spark session
  spark = SparkSession.builder.appName("streamer").getOrCreate()
  
  # create DataFrame representing the stream
  df = spark.readStream \
    .format("kafka").option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "mytopic") \
    .load() \
    .selectExpr("Timestamp", "cast (value as string) as json") \
    .select("Timestamp", F.from_json("json", schema_simple).alias('json_wrapper')) \
    .selectExpr("Timestamp", "json_wrapper.Signal", "json_wrapper.Value")
  df.printSchema()
  # Do some dummy processing
  df2 = df.filter("Value &lt; 11111111111")
  print("df2: ", df2.isStreaming)
  
  def process_row(row):
      # Process row
      row.write.jdbc(url=db_target_url, table="mytopic", mode="append", properties=db_target_properties)
      pass
  query = df2.writeStream.foreach(process_row).start()
  </code></pre>
  
  <p>I get an error:</p>
  
  <blockquote>
    <p>AttributeError: write</p>
  </blockquote>
  
  <p>Why?</p>