如何在 PySpark 中使用 foreach 或 foreachBatch 写入数据库？

Question

我想使用 Python (PySpark) 从 Kafka 源到 MariaDB 执行 Spark Structured Streaming (Spark 2.4.x)。

我想使用流式 Spark 数据帧，而不是静态或 Pandas 数据帧。

根据https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks，似乎必须使用foreach或foreachBatch因为流式数据帧没有可能的数据库接收器。

这是我的尝试：

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 < 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()

我收到一个错误：

属性错误：写

为什么？

Answer 1

tl;dr替换foreach为foreachBatch.

引用官方文档：

在foreach与foreachBatch操作让您应用在流媒体查询的输出任意操作和写作的逻辑。它们的用例略有不同——虽然foreach允许在每一行上自定义写入逻辑，但foreachBatch允许对每个微批次的输出进行任意操作和自定义逻辑。

换句话说，您writeStream.foreach(process_row)对没有write.jdbc可用的单行（数据）进行操作，因此出现错误。

将行视为一段数据，您可以使用任何您想要的 API 将其保存在您想要的任何位置。

如果您确实需要 Spark 的支持（并且确实使用write.jdbc），则实际上应该使用foreachBatch.

whileforeach允许在每一行上自定义写入逻辑，foreachBatch允许对每个 micro-batch 的输出进行任意操作和自定义逻辑。

Answer 2

在 Jacek 的支持下，我可以修复我的示例：

def process_row(df, epoch_id):
    df2.write.jdbc(url=db_target_url, table="mytopic", mode="append", properties=db_target_properties)
    pass
query = df2.writeStream.foreachBatch(process_row).start()

您还必须将epoch_id放入函数参数中。否则，您会在 Spark 日志文件中收到 Jupyter Notebook 中未显示的错误。