尝试使用Kafka和pyspark在postgreSQL中从spark编写流式数据帧

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType

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

# 定义Kafka主题和postgreSQL连接信息
kafka_topic = "your_kafka_topic"
kafka_bootstrap_servers = "your_kafka_bootstrap_servers"
postgres_url = "your_postgres_url"
postgres_table = "your_postgres_table"

# 定义流式数据帧的模式
schema = StructType([
    StructField("field1", StringType(), True),
    StructField("field2", StringType(), True),
    ...
])

# 从Kafka读取数据
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", kafka_topic) \
    .load()

# 解析JSON数据
parsed_df = df.selectExpr("CAST(value AS STRING)") \
    .select(from_json("value", schema).alias("data")) \
    .select("data.*")

# 将数据写入postgreSQL
query = parsed_df \
    .writeStream \
    .format("jdbc") \
    .option("url", postgres_url) \
    .option("dbtable", postgres_table) \
    .option("user", "your_postgres_username") \
    .option("password", "your_postgres_password") \
    .start()

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