如何使用MSK Connect将AWS MSK(Kafka)数据流式传输到Snowflake

前置准备 #

• 提前在Snowflake侧完成服务账号、数据库、Schema、虚拟仓库、外部阶段、管道的创建，给服务账号授予对应资源的读写权限，该步骤和EC2上部署Snowflake Kafka连接器的前置操作完全一致。
• 确认你的MSK集群版本支持MSK Connect功能，集群网络配置允许MSK连接器访问。

步骤1：上传连接器插件到S3 #

下载兼容Kafka 2.8.x及以上版本的Snowflake Kafka连接器压缩包，解压后将所有jar包和依赖文件打包为zip格式，上传到你指定的S3存储桶路径，同时给MSK Connect服务角色授予该S3路径的读取权限。

步骤2：创建MSK自定义插件 #

进入AWS MSK控制台，选择「自定义插件」→「创建自定义插件」，选择你刚才上传到S3的连接器zip包，填写插件名称完成创建，等待插件状态变更为可用。

步骤3：配置连接器核心属性 #

创建连接器时，核心配置参数参考如下，按需修改对应取值即可：

# 通用连接器配置
connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
tasks.max=适配Kafka主题分区数的任务数
topics=需要同步的Kafka主题名称
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

# Snowflake对接配置
snowflake.url.name=你的Snowflake访问URL，格式为<账户标识符>.snowflakecomputing.com:443
snowflake.user.name=Snowflake服务账号用户名
snowflake.private.key=服务账号私钥明文，需去掉PEM格式头尾标记和所有换行符
snowflake.private.key.passphrase=私钥加密口令，无加密可删除该配置
snowflake.database.name=目标Snowflake数据库名
snowflake.schema.name=目标Snowflake Schema名
snowflake.role.name=服务账号绑定的角色名
snowflake.warehouse.name=使用的Snowflake虚拟仓库名
snowflake.stage.name=提前创建的外部阶段名，格式为<数据库>.<Schema>.<阶段名>
snowflake.pipe.name=提前创建的管道名，格式为<数据库>.<Schema>.<管道名>

# 批量写入优化配置
snowflake.buffer.count.records=批量写入记录数阈值，默认10000
snowflake.buffer.flush.time=批量刷新时间阈值，单位秒，默认10
snowflake.buffer.size.bytes=批量写入字节阈值，默认20000000

步骤4：配置访问权限与网络 #

• 给连接器执行角色授予MSK主题读写权限、CloudWatch日志写入权限
• 如果MSK集群部署在私有VPC，连接器需要选择相同VPC下的子网和有权限访问MSK集群的安全组，同时确保该网络环境可以正常访问Snowflake服务。

常见排障点 #

• 鉴权失败优先检查私钥格式是否正确、Snowflake服务账号权限是否足够
• 连接超时优先检查VPC网络、安全组、NACL是否放行Snowflake访问流量
• 数据写入失败优先检查Snowflake管道、阶段配置是否正确，Kafka消息格式是否和转换器配置匹配

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