Debezium Oracle连接器：如何忽略特定列生成Kafka消息并同步至Snowflake？

一、源端Debezium Oracle连接器直接排除列 #

Debezium Oracle连接器本身提供column.exclude.list参数，可直接在源连接器配置中指定需要排除的BLOB列，从源头避免不必要的数据捕获与传输，无需依赖Kafka Connect转换。配置示例：

connector.class=io.debezium.connector.oracle.OracleConnector
# 其他基础配置（如数据库连接、topic前缀等）...
column.exclude.list=schema_name.table_a.blob_col,schema_name.table_b.blob_col

若多表的BLOB列名一致，可使用通配符简化配置：

column.exclude.list=*.blob_col

二、修正Kafka Connect ReplaceField转换配置 #

你之前尝试的ReplaceField未生效，核心原因是Debezium输出的消息为嵌套结构（包含before/after/source等层级），直接对Value层级过滤无法命中after中的列。需结合ExtractNewRecordState先展开数据，再进行字段过滤：

transforms=extract,removeBlob
# 提取新记录的有效数据（将after中的内容作为消息主体）
transforms.extract.type=io.debezium.transforms.ExtractNewRecordState
transforms.extract.drop.tombstones=false

# 排除指定BLOB列
transforms.removeBlob.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.removeBlob.blacklist=blob_col

若多表BLOB列名不同，需逐一添加到blacklist；若仍存在嵌套结构，需调整转换的作用层级（如针对嵌套字段路径配置）。

三、Sink端Snowflake侧处理 #

若源端和Kafka转换配置受限，可在Snowflake端完成列过滤：

• 预创建不含BLOB列的目标表：手动在Snowflake中创建与Oracle表结构匹配但剔除BLOB列的表，关闭Snowflake Sink连接器的table.auto.create和table.auto.evolve参数，指定连接器写入预定义表：

  connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
  # 其他基础配置（如账户、仓库等）...
  table.auto.create=false
  table.auto.evolve=false
  snowflake.topic2table.map=your_topic:target_schema.target_table
  

• 使用Snowpipe COPY语句过滤列：若通过Snowpipe加载数据，可在COPY命令中指定排除BLOB列：

  COPY INTO target_schema.target_table
  FROM @your_stage
  FILE_FORMAT = (TYPE = JSON)
  EXCLUDE_COLUMNS = ('blob_col');
  

• 配置Snowflake Sink列白名单：利用连接器的columns.include.list参数，列出所有需要同步的列（间接排除BLOB列）：

  columns.include.list=col1,col2,col3  # 仅同步指定列，不包含BLOB列
  

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