Flink SQL写入DynamoDB报错：提供的键元素与schema不匹配

报错信息 #

Caused by: software.amazon.awssdk.services.dynamodb.model.DynamoDbException: The provided key element does not match the schema (Service: DynamoDb, Status Code: 400, Request ID: ...)

数据校验正常，restaurant_id值有效，Sink前的print输出也显示该字段无问题，类型已对齐为字符串。

DynamoDB表结构 #

主表主键为简单分区键restaurant_id（字符串类型），无排序键，仅GSI使用复合键，任务仅写入主表：

dynamodb:
  OnlineFeatures:
    attributes:
        - name: "restaurant_id"
          type: "S"
        ...
    hash_key: "restaurant_id"
    # 主表无排序键
    global_secondary_indexes:
      - name: "country_index"
        hash_key: "_country"
        range_key: "_training_timestamp"
        projection_type: "ALL"

Flink SQL代码 #

已将restaurant_id转换为STRING类型，表定义和插入语句如下：

-- DynamoDB Sink表定义
CREATE TABLE dynamodb_sink_table (
    _window_end STRING,
    ...
    restaurant_id STRING,
    ...
) PARTITIONED BY ( restaurant_id )
WITH (
    'connector' = 'dynamodb',
    'table-name' = '{dynamodb_table_name}',
    'aws.region' = '{config.aws_region}',
    'sink.ignore-nulls' = 'true'
);

-- 数据写入语句
INSERT INTO dynamodb_sink_table
SELECT 
    ...,
    CAST(dfr.delco_restaurant_id AS STRING) as restaurant_id,
    ...
FROM df_rests dfr 
LEFT JOIN ...
WHERE dfr.restaurant_id IS NOT NULL;

核心疑问 #

Flink任务处理数据正常，但DynamoDB提示键结构不匹配而非数据问题：

1. 缺少哪些配置？
2. Flink的什么行为会导致发送的键结构与主表简单分区键不匹配？
3. 如何强制Flink仅用restaurant_id作为写入请求的键？

问题根源 #

Flink DynamoDB连接器未明确主键配置时，会自动推断主键逻辑：若表使用PARTITIONED BY，可能默认将分区字段作为主键，但如果存在与GSI键名重合的字段，连接器可能误将这些字段加入主键结构，导致发送给DynamoDB的请求包含多余键元素，触发结构不匹配报错。

具体修复步骤 #

1. 明确指定DynamoDB主键
   在Flink表的WITH参数中添加分区键配置，强制连接器仅使用restaurant_id作为主键：

CREATE TABLE dynamodb_sink_table (
    _window_end STRING,
    ...
    restaurant_id STRING,
    ...
) PARTITIONED BY ( restaurant_id )
WITH (
    'connector' = 'dynamodb',
    'table-name' = '{dynamodb_table_name}',
    'aws.region' = '{config.aws_region}',
    'sink.ignore-nulls' = 'true',
    'dynamodb.primary-key.hash-key-name' = 'restaurant_id' -- 明确分区键
);

此配置会覆盖自动推断逻辑，确保仅向DynamoDB发送restaurant_id作为主键。

2. 禁用排序键配置
   由于主表无排序键，确保WITH参数中未设置dynamodb.primary-key.range-key-name，避免连接器添加多余的排序键元素。

3. 确认主键非空
   虽然SQL中已过滤空值，但需最终验证写入sink_table的restaurant_id无空值——DynamoDB不接受空主键，即使开启sink.ignore-nulls也无效。

4. 升级连接器版本
   若使用旧版Flink DynamoDB连接器，可能存在主键推断bug，建议升级到Flink 1.17+对应的稳定版连接器。

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