EMR上Flink写入S3 Iceberg异常：数据损坏+计数不符

在EMR上使用Flink，Kafka部署于EKS集群，二者连通正常。尝试通过Table API将Kafka Topic的数据写入S3上的Iceberg表，数据虽已写入但存在损坏情况；且Flink UI显示仅摄入1条记录，却统计出950+条处理量，全程无任何异常报错。

CREATE OR REPLACE TABLE kafka_source (  
  id    INT,  
  data  STRING  
) WITH (  
  'connector'                    = 'kafka',  
  'topic'                        = 'test-topic',  
  'properties.bootstrap.servers' = 'something.us-east-1.elb.amazonaws.com:9094',  
  'properties.group.id'          = 'flink-iceberg-consumer',  
  'scan.startup.mode'            = 'earliest-offset',  
  'format'                       = 'json'  
);  

CREATE CATALOG glue_catalog WITH (  
  'type'         = 'iceberg',  
  'warehouse'    = 's3://eks-benchmark-iceberg/warehouse/',  
  'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog',  
  'io-impl'      = 'org.apache.iceberg.aws.s3.S3FileIO' 
);  

USE CATALOG glue_catalog;  

CREATE DATABASE IF NOT EXISTS benchmark_db;  

USE benchmark_db;  

CREATE TABLE IF NOT EXISTS orders_iceberg (  
  id    INT,  
  data  STRING  
);  

USE CATALOG default_catalog;
SET 'execution.runtime-mode' = 'streaming';
ADD JAR '/usr/lib/flink/lib/flink-sql-connector-kafka-3.3.0-1.20.jar';  
INSERT INTO glue_catalog.benchmark_db.orders_iceberg 
SELECT id, data FROM kafka_source; 

{"id": 1, "data": "hello"}

• Flink version: 1.20.0
• Kafka version: 4.x

1. 补全Iceberg表核心配置 #

创建Iceberg表时未指定存储格式与流式提交规则，这是数据损坏的常见原因。修改建表语句，补充必要配置：

CREATE TABLE IF NOT EXISTS orders_iceberg (  
  id    INT,  
  data  STRING  
) WITH (
  'format' = 'parquet', -- 指定标准列式存储格式，避免无格式写入导致损坏
  'write.commit.interval-ms' = '10000', -- 流式场景下固定提交间隔，避免频繁小文件
  'write.distribution-mode' = 'hash', -- 按字段哈希分区写入，保证数据分布均匀
  'write.metadata.delete-after-commit.enabled' = 'true' -- 清理过期元数据，避免元数据混乱
);

2. 严格Kafka JSON格式解析规则 #

虽然指定了format='json'，但未配置解析异常处理，隐性的解析错误可能导致数据重复处理或损坏。修改Kafka源表配置：

ALTER TABLE kafka_source SET (
  'json.fail-on-missing-field' = 'true', -- 字段缺失时直接报错，避免脏数据流入
  'json.ignore-parse-errors' = 'false' -- 解析失败时终止任务，快速定位问题
);

3. 排查Flink任务重复重启问题 #

处理量远大于摄入记录，大概率是任务隐性重启导致重复消费。检查以下配置：

• 确认Flink的state.backend已配置为S3（而非内存），防止状态丢失导致任务回溯重复处理
• 查看JobManager日志，检查是否存在TaskManager内存不足、心跳超时等重启事件
• 验证Kafka消费者的auto.offset.reset未被EMR集群配置覆盖，确保scan.startup.mode='earliest-offset'仅生效一次

4. 确认依赖兼容性 #

Flink 1.20.0需要匹配对应版本的Iceberg连接器，EMR默认安装的Iceberg版本可能存在兼容问题。确保集群中flink-sql-connector-iceberg版本与Flink 1.20.0适配，避免依赖冲突导致数据写入逻辑异常。

5. 处理S3最终一致性问题 #

S3的最终一致性可能导致写入后立即读取看到损坏数据，建议等待5-10分钟后再验证数据完整性。同时，若EMR集群角色权限不足，需在Glue Catalog中显式配置S3访问密钥：

ALTER CATALOG glue_catalog SET (
  's3.access-key-id' = 'your-access-key',
  's3.secret-access-key' = 'your-secret-key'
);

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