https://stackoverflow.com/questions/58766638/how-to-use-foreach-or-foreachbatch-in-pyspark-to-write-to-database
    尚未试验过，走读觉得应该可以
    注：foreach和foreachBatch是用于处理流数据的

    1. from pyspark.sql import SparkSession
    2. import pyspark.sql.functions as F
    3. from pyspark.sql.types import StructField, StructType, StringType, DoubleType, TimestampType
    4. from pyspark.sql import DataFrameWriter
    5. # configuration of target db
    6. db_target_url = "jdbc:mysql://localhost/database"
    7. db_target_properties = {"user":"writer", "password":"1234"}
    8. # schema
    9. schema_simple = StructType([StructField("Signal", StringType()),StructField("Value", DoubleType())])
    11. # create spark session
    12. spark = SparkSession.builder.appName("streamer").getOrCreate()
    14. # create DataFrame representing the stream
    15. df = spark.readStream \
    16.   .format("kafka").option("kafka.bootstrap.servers", "localhost:9092") \
    17.   .option("subscribe", "mytopic") \
    18.   .load() \
    19.   .selectExpr("Timestamp", "cast (value as string) as json") \
    20.   .select("Timestamp", F.from_json("json", schema_simple).alias('json_wrapper')) \
    21.   .selectExpr("Timestamp", "json_wrapper.Signal", "json_wrapper.Value")
    22. df.printSchema()
    23. # Do some dummy processing
    24. df2 = df.filter("Value < 11111111111")
    25. print("df2: ", df2.isStreaming)
    27. def process_row(row):
    28.     # Process row
    29.     row.write.jdbc(url=db_target_url, table="mytopic", mode="append", properties=db_target_properties)
    30.     pass
    31. query = df2.writeStream.foreachBatch(process_row).start()