Snowflake Kafka Sink Connector运行正常但无数据写入问题求助

• 检查Kafka主题是否有新数据

  • 使用Kafka命令行工具查看消费者组的消费状态，确认偏移量是否增长：

    kafka-consumer-groups.sh --describe --group connect-SF_2024-05-13_REFERENCES_ABS --bootstrap-server <你的Kafka集群地址>
    

  • 手动消费主题数据，验证是否有消息产生：

    kafka-console-consumer.sh --topic <目标主题名> --bootstrap-server <你的Kafka集群地址> --from-beginning
    

• 验证Snowflake管道状态与加载历史

  • 在Snowflake中查询管道实时状态，查看是否存在待加载或失败的文件：

    SELECT SYSTEM$PIPE_STATUS('DEV_ENTERPRISE_DATALAKE_RAW.DEV_CRM_KAFKA_FINAL_TOPIC_RAW.SNOWFLAKE_KAFKA_CONNECTOR_SF_2024_05_13_REFERENCES_ABS_533516335_PIPE_REFERENCES_ABS_5');
    

  • 查看管道的加载历史记录，排查是否有加载错误：

    SELECT * FROM TABLE(INFORMATION_SCHEMA.PIPE_LOAD_HISTORY(PIPE_NAME => 'DEV_ENTERPRISE_DATALAKE_RAW.DEV_CRM_KAFKA_FINAL_TOPIC_RAW.SNOWFLAKE_KAFKA_CONNECTOR_SF_2024_05_13_REFERENCES_ABS_533516335_PIPE_REFERENCES_ABS_5'));
    

• 检查连接器的消费偏移量配置

  • 确认连接器配置中的auto.offset.reset参数：若设置为latest，连接器只会消费部署后的新消息；如需消费历史数据，改为earliest后重新部署。
  • 检查消费者组是否已提交过偏移量，导致连接器从最新位置开始消费，无历史数据可拉取。

• 排查网络连接稳定性

  • 日志中Node -1 disconnected提示Kafka客户端与集群连接存在波动，检查Connector所在节点与Kafka集群的网络连通性，确认Kafka集群状态正常。
  • 验证Snowflake Kafka Connector的Kafka客户端版本与集群版本兼容，避免版本不匹配引发的消费异常。

• 校验数据格式与表结构兼容性

  • 确认Kafka消息格式（JSON/Avro等）与Snowflake表的字段类型、数量匹配，不存在字段缺失或类型不兼容的情况。
  • 若使用Schema Registry，检查Schema是否正确注册，且与Snowflake表结构一致。

• 检查Snowflake内部阶段存储并手动触发加载

  • 查看表关联的内部阶段是否有未被管道处理的文件：

    LIST @%DEV_ENTERPRISE_DATALAKE_RAW.DEV_CRM_KAFKA_FINAL_TOPIC_RAW.<目标表名>;
    

  • 若存在未处理文件，手动触发管道加载：

    ALTER PIPE DEV_ENTERPRISE_DATALAKE_RAW.DEV_CRM_KAFKA_FINAL_TOPIC_RAW.SNOWFLAKE_KAFKA_CONNECTOR_SF_2024_05_13_REFERENCES_ABS_533516335_PIPE_REFERENCES_ABS_5 REFRESH;
    

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