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

2024-05-05 • 问答

我想使用Python（PySpark）从Kafka源到MariaDB进行Spark结构化流（Spark 2.4.x）。

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

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

这是我的尝试：

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

我得到一个错误：

AttributeError：写入

为什么？

tl; dr 将UPDATE myTable SET myGeo = NULL WHERE ST_IsEmpty(myGeo) IS NULL;替换为<?php foreach ($departments['choices'] as $key => $department) : ?> <input type="checkbox" name="<?= $department ?>" id="<?= $department ?>" value="unchecked" />。

引用official documentation：

通过foreach和foreachBatch操作，您可以在流查询的输出上应用任意操作并编写逻辑。它们的用例略有不同-foreach允许在每一行上进行自定义写逻辑，而foreachBatch允许在每个微批处理的输出上进行任意操作和自定义逻辑。

换句话说，您的foreach对没有foreachBatch可用的一行（数据）进行操作，因此会出错。

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

如果您确实需要Spark的支持（并使用writeStream.foreach(process_row)），则应实际使用write.jdbc。

write.jdbc允许在每一行上进行自定义写逻辑，而foreachBatch允许在每个微批处理的输出上进行任意操作和自定义逻辑。

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

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

大家都在问