如何配置本地环境从VPC内的AWS MSK集群消费数据？

第一步：打通本地与MSK的网络 #

MSK集群部署在VPC内，本地默认无法直接访问，任选以下一种方式解决网络连通问题：

• AWS Client VPN：在AWS控制台创建Client VPN端点，下载配置文件后通过AWS VPN Client连接，确保本地能ping通MSK Broker的私有IP。
• SSH隧道转发：借助VPC内的EC2实例作为跳板，执行命令将MSK端口映射到本地：

  ssh -i "你的密钥对文件.pem" -L 9092:<MSK Broker私有IP>:9092 ec2-user@<EC2公网IP>
  

  完成后本地可通过localhost:9092访问MSK集群。

第二步：调整Spark Kafka消费代码 #

根据MSK集群的认证方式，补全代码配置：

IAM认证（AWS官方推荐） #

streaming_query = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "填写MSK Broker私有IP或本地转发的localhost:9092") \
    .option("subscribe", input_topic) \
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism", "AWS_MSK_IAM") \
    .option("kafka.sasl.jaas.config", "software.amazon.msk.auth.iam.IAMLoginModule required;") \
    .option("kafka.sasl.client.callback.handler.class", "software.amazon.msk.auth.iam.IAMClientCallbackHandler") \
    .load()

本地需配置AWS凭证：可在~/.aws/credentials文件中填写密钥，或设置AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY环境变量，同时确保账号拥有MSK的kafka:DescribeCluster和kafka:ReadData权限。

SCRAM认证 #

streaming_query = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "填写MSK Broker私有IP") \
    .option("subscribe", input_topic) \
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism", "SCRAM-SHA-512") \
    .option("kafka.sasl.jaas.config", 'org.apache.kafka.common.security.scram.ScramLoginModule required username="你的SCRAM用户名" password="你的SCRAM密码";') \
    .load()

无认证（仅测试场景使用，禁止生产环境） #

网络连通后直接填写MSK Broker私有IP即可：

streaming_query = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "填写MSK Broker私有IP") \
    .option("subscribe", input_topic) \
    .load()

本地测试关键注意事项 #

• 依赖包需齐全：使用IAM认证时，启动Spark需添加依赖参数--packages software.amazon.msk:aws-msk-iam-auth:1.1.5。
• 先验证网络连通性：用Kafka自带控制台消费者测试，比如隧道转发后执行：

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic <你的主题名> --from-beginning
  

• 安全组配置：确保MSK集群的安全组允许来自本地VPN网段或EC2跳板实例的9092/9094端口入站流量。

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