如何使用ForeachWriter在Python语言中使用结构化火花流在Mongodb中插入行？

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.streaming import StreamingQuery
from pymongo import MongoClient

# 创建SparkSession
spark = SparkSession.builder \
    .appName("StructuredStreamingWithMongoDB") \
    .getOrCreate()

# 创建MongoDB连接
mongo_client = MongoClient("mongodb://localhost:27017/")
db = mongo_client["mydb"]
collection = db["mycollection"]

# 定义ForeachWriter类
class MongoForeachWriter:
    def open(self, partition_id, epoch_id):
        # 在此处打开MongoDB连接
        self.client = MongoClient("mongodb://localhost:27017/")
        self.db = self.client["mydb"]
        self.collection = self.db["mycollection"]
        return True

    def process(self, row):
        # 在此处处理每一行数据，并将其插入MongoDB
        document = row.asDict()
        self.collection.insert_one(document)

    def close(self, error):
        # 在此处关闭MongoDB连接
        self.client.close()

# 读取结构化流数据
stream_data = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "mytopic") \
    .load()

# 对流数据进行处理
processed_data = stream_data.selectExpr("CAST(value AS STRING)")

# 创建ForeachWriter实例
mongo_writer = MongoForeachWriter()

# 将数据写入MongoDB
query = processed_data.writeStream \
    .foreach(mongo_writer) \
    .start()

# 等待流处理完成
query.awaitTermination()