无主键Postgres表经Kafka同步至BigQuery报错的解决方案咨询

当前数据流链路为：Postgres → Kafka → BigQuery。带主键的Postgres表对应的Kafka主题包含Avro键和值Schema，这类表可正常同步至BigQuery；但无主键的表对应的Kafka主题没有Avro键Schema，创建BigQuery Sink Connector时会遇到两类错误：

• 保留key.converter配置时，报错：Caused by: com.wepay.kafka.connect.bigquery.exception.ConversionConnectException: Only Map objects supported in absence of schema for record conversion to BigQuery format.
• 移除key.converter配置时，触发错误：“Top-level Kafka Connect schema must be of type 'struct'”

原连接器配置如下：

{"project": "staging","defaultDataset": "data_lake","keyfile": "<redacted>","keySource": "JSON","sanitizeTopics": "true","kafkaKeyFieldName": "_kid","autoCreateTables": "true","allowNewBigQueryFields": "true","upsertEnabled": "false","bigQueryRetry": "5","bigQueryRetryWait": "120000","bigQueryPartitionDecorator": "false","name": "hd-sink-bq","connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector","tasks.max": "1","key.converter": "io.confluent.connect.avro.AvroConverter","key.converter.schema.registry.url": "<redacted>","key.converter.basic.auth.credentials.source": "USER_INFO","key.converter.schema.registry.basic.auth.user.info": "<redacted>","value.converter": "io.confluent.connect.avro.AvroConverter","value.converter.schema.registry.url": "<redacted>","value.converter.basic.auth.credentials.source": "USER_INFO","value.converter.schema.registry.basic.auth.user.info": "<redacted>","topics": "public.event_issues","errors.tolerance": "all","errors.log.include.messages": "true","errors.deadletterqueue.topic.name": "connect.bq-sink.deadletter","errors.deadletterqueue.topic.replication.factor": "1","errors.deadletterqueue.context.headers.enable": "true","transforms": "tombstoneHandler","offset.flush.timeout.ms": "300000","transforms.dropNullRecords.predicate": "isNullRecord","transforms.dropNullRecords.type": "org.apache.kafka.connect.transforms.Filter","transforms.tombstoneHandler.behavior": "drop_warn","transforms.tombstoneHandler.type": "io.aiven.kafka.connect.transforms.TombstoneHandler"}

方案1：调整键转换器并添加Transform处理键格式 #

针对无主键表的Kafka主题，替换键转换器并添加Transform生成符合要求的键结构：

• 将key.converter替换为JSON转换器，同时禁用键的Schema校验：

  key.converter=org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable=false
  

• 添加InsertField Transform，将键转换为BigQuery支持的Map格式：

  transforms=insertKey,tombstoneHandler
  transforms.insertKey.type=org.apache.kafka.connect.transforms.InsertField$Key
  transforms.insertKey.static.field=_dummy_key
  transforms.insertKey.static.value=default
  

  该配置会为每条记录的键添加固定字段，将无Schema的键转为Map结构，满足连接器的格式要求。

方案2：从Postgres源端生成有效键 #

如果使用Debezium作为Postgres源连接器，可为无主键表指定唯一标识字段作为键，确保Kafka主题的键带有Avro Schema：

• 在Debezium源连接器配置中添加：

  table.include.list=public.event_issues
  message.key.columns=public.event_issues:column1,column2
  

  将column1,column2替换为目标表中可唯一标识记录的字段（支持组合字段），这样Kafka主题的键会自动带上Avro Schema，无需修改Sink端的key.converter配置即可正常同步。

调整后的Sink连接器配置示例 #

{"project": "staging","defaultDataset": "data_lake","keyfile": "<redacted>","keySource": "JSON","sanitizeTopics": "true","kafkaKeyFieldName": "_kid","autoCreateTables": "true","allowNewBigQueryFields": "true","upsertEnabled": "false","bigQueryRetry": "5","bigQueryRetryWait": "120000","bigQueryPartitionDecorator": "false","name": "hd-sink-bq","connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector","tasks.max": "1","key.converter": "org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": "false","value.converter": "io.confluent.connect.avro.AvroConverter","value.converter.schema.registry.url": "<redacted>","value.converter.basic.auth.credentials.source": "USER_INFO","value.converter.schema.registry.basic.auth.user.info": "<redacted>","topics": "public.event_issues","errors.tolerance": "all","errors.log.include.messages": "true","errors.deadletterqueue.topic.name": "connect.bq-sink.deadletter","errors.deadletterqueue.topic.replication.factor": "1","errors.deadletterqueue.context.headers.enable": "true","transforms": "insertKey,tombstoneHandler","offset.flush.timeout.ms": "300000","transforms.insertKey.type": "org.apache.kafka.connect.transforms.InsertField$Key","transforms.insertKey.static.field": "_dummy_key","transforms.insertKey.static.value": "default","transforms.tombstoneHandler.behavior": "drop_warn","transforms.tombstoneHandler.type": "io.aiven.kafka.connect.transforms.TombstoneHandler"}

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