如何使用PySpark将Kafka消息值转换为指定Schema？

你可以通过展开from_json返回的Struct类型字段得到顶级Schema，两种常用实现方式如下：

方法1：自动展开Struct所有子字段 #

先为解析后的Struct列设置别名，再用.*匹配所有子字段：

df = spark.read.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test1").load()
userSchema = StructType().add("Name", StringType(), True).add("Age", IntegerType(), True)
df1 = df.selectExpr("CAST(value AS STRING)")
# 先给解析结果起别名，再展开所有子字段
df2 = df1.select(from_json(col("value"), userSchema).alias("parsed_data")).select("parsed_data.*")

df2.printSchema()

输出结果和预期一致：

root
 |-- Name: string (nullable = true)
 |-- Age: integer (nullable = true)

方法2：手动指定提取字段 #

如果需要明确控制输出字段、或者只需要部分子字段，可以手动指定提取：

df2 = df1.select(
    from_json(col("value"), userSchema)["Name"].alias("Name"),
    from_json(col("value"), userSchema)["Age"].alias("Age")
)

内容的提问来源于stack exchange，提问作者Rishabh Sairawat