Structured Streaming写入Strimzi Kafka突发故障排查求助

一、证书路径的正确访问方式 #

1. 确认提交命令的正确性 #

使用--files参数传递GCS证书时，需确保路径格式正确，多文件用逗号分隔（无空格）：

gcloud dataproc jobs submit pyspark \
--cluster=your-cluster-id \
--files=gs://your-bucket/ssl/ca.pem,gs://your-bucket/ssl/client-cert.pem,gs://your-bucket/ssl/client-key.pem \
your_streaming_script.py

验证GCS文件存在性：

gsutil ls gs://your-bucket/ssl/ca.pem

2. 代码中正确获取证书路径 #

使用SparkFiles.get()获取同步到Executor节点的证书绝对路径，注意需在Driver端配置Kafka参数时使用该路径（Spark会自动同步文件到所有Executor）：

from pyspark import SparkFiles
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("KafkaStreamTransfer").getOrCreate()

# 获取证书绝对路径
ca_path = SparkFiles.get("ca.pem")
cert_path = SparkFiles.get("client-cert.pem")
key_path = SparkFiles.get("client-key.pem")

# 配置Kafka写入参数
kafka_write_conf = {
    "kafka.bootstrap.servers": "gke-kafka-broker:9093",
    "kafka.security.protocol": "SSL",
    "kafka.ssl.ca.location": ca_path,
    "kafka.ssl.certificate.location": cert_path,
    "kafka.ssl.key.location": key_path,
    "kafka.ssl.key.password": "your-key-passphrase",
    "subscribe": "target-topic",
    "checkpointLocation": "gs://your-bucket/checkpoint-dir"
}

# 读取源Kafka并写入目标Kafka
source_df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "vm-kafka-broker:9092").option("subscribe", "source-topic").load()
source_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
    .writeStream.format("kafka").options(**kafka_write_conf).start().awaitTermination()

3. 排查SparkFiles.get()失效的原因 #

• 检查Driver日志：查看作业日志中SparkFiles.get()的返回路径，确认文件是否真的不存在（日志路径可在Dataproc作业详情页获取）。
• 避免在UDF中调用SparkFiles.get()：UDF运行在Executor端，若需在UDF中访问证书，需确保文件已同步，或在Driver端提前获取路径并传递给UDF。
• 验证Executor节点文件存在性：登录Worker节点，检查SparkFiles.get()返回的路径（如/tmp/spark-xxxx/files/ca.pem）是否存在。

二、TimeoutException（元数据获取超时）排查 #

1. 网络连通性验证 #

在Dataproc Worker节点上，用原生Kafka客户端测试连接（避免本地测试的网络差异）：

# 安装kafka-python
pip install kafka-python

# 测试生产消息
python3 -c "
from kafka import KafkaProducer
producer = KafkaProducer(
    bootstrap_servers=['gke-kafka-broker:9093'],
    security_protocol='SSL',
    ssl_cafile='/hadoop/yarn/local/usercache/root/appcache/application_xxxx/container_xxxx/files/ca.pem',
    ssl_certfile='/hadoop/yarn/local/usercache/root/appcache/application_xxxx/container_xxxx/files/client-cert.pem',
    ssl_keyfile='/hadoop/yarn/local/usercache/root/appcache/application_xxxx/container_xxxx/files/client-key.pem',
    ssl_password='your-key-passphrase'
)
producer.send('test-topic', b'test-message')
producer.flush()
print('Connection successful')
"

• 若测试失败：排查GKE Kafka集群的防火墙规则（是否允许Dataproc节点IP访问9093端口）、VPC peering配置、Broker状态。
• 若测试成功：问题出在Spark与Kafka客户端的兼容性或代码配置。

2. 版本兼容性检查 #

确认Dataproc Spark版本对应的Kafka客户端版本与GKE Kafka集群版本兼容（如Spark 3.3对应kafka-clients 3.2.0，需与GKE Kafka版本差距不超过1个大版本）。若近期升级过Dataproc或Kafka集群，需回退验证。

3. 其他可能原因 #

• 证书过期：虽然本地测试正常，但需确认Dataproc上传递的证书是否为最新有效版本。
• Kafka Broker负载过高：查看GKE Kafka集群的监控指标（如元数据请求延迟、Broker CPU/内存使用率），排查是否存在Broker故障或资源瓶颈。

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