如何在PySpark/Databricks中正确解码Pub/Sub消息的Payload？

你遇到的问题核心是消息解码与JSON解析的步骤有误，或是对Pub/Sub消息的编码格式判断偏差。以下是修正后的完整处理流程：

1. 正确解码消息Payload #

Pub/Sub的payload字段默认是base64编码的二进制数据，但如果消息发送时未做base64编码，强行调用unbase64会直接导致数据损坏（就是你看到的乱码或无格式字符串）。

情况1：消息未做base64编码 #

直接解码payload为UTF-8字符串：

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

# 读取Pub/Sub流
df = (spark.readStream.format("pubsub")
    .option("subscriptionId", SUBSCRIPTION_ID)
    .option("projectId", PROJECT_ID)
    .option("serviceCredential", SERVICE_CREDENTIAL)
    .option("topicId", TOPIC_ID)
    .load())

# 直接解码payload为JSON字符串
df = df.withColumn("json_str", decode(col("payload"), "UTF-8"))

情况2：消息确实是base64编码 #

先做base64解码，再转UTF-8字符串：

from pyspark.sql.functions import unbase64, decode, from_json, col

# 先解码base64，再转UTF-8字符串
df = df.withColumn("decoded_payload", unbase64(col("payload")))
df = df.withColumn("json_str", decode(col("decoded_payload"), "UTF-8"))

2. 将JSON字符串解析为结构化数据 #

拿到合法的JSON字符串后，定义匹配消息结构的Schema，用from_json解析为结构化列：

# 定义与消息格式对应的Schema
event_schema = StructType([
    StructField("event_id", StringType(), nullable=True),
    StructField("user_id", StringType(), nullable=True),
    StructField("session_id", StringType(), nullable=True),
    StructField("browser", StringType(), nullable=True),
    StructField("uri", StringType(), nullable=True),
    StructField("event_type", StringType(), nullable=True)
])

# 解析JSON字符串为结构化数据
df = df.withColumn("event_data", from_json(col("json_str"), event_schema))

# 展开结构化列到顶层（可选，方便写入Delta表）
df = df.select("event_data.*")

3. 写入Delta表 #

将结构化后的流数据写入Delta表，注意必须指定检查点路径：

query = (df.writeStream
    .format("delta")
    .option("checkpointLocation", "/your/checkpoint/path")
    .table("target_delta_table"))

query.awaitTermination()

额外排查要点 #

• 如果仍出现乱码，检查消息发送方的编码格式：若使用了UTF-8以外的编码（如GBK），将decode的第二个参数替换为对应编码。
• 直接在Pub/Sub控制台查看原始消息内容，确认发送方是否确实发送了标准JSON格式的Payload。

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