PAYATHON 2026

Flink 写入 S3 Iceberg 表后无法读取，且处理记录数异常，如何排查？

数据小许

结论

现有信息还不能证明 Iceberg 数据已经损坏，也无法确定超过 950 的计数代表 950 条业务记录。建议先核对读取的文件格式、Kafka 的实际消费范围，以及 Iceberg 快照是否成功提交。

示例 SQL 中，scan.startup.mode 设置为 earliest-offset，但没有展示 checkpoint 配置。EMR 或集群可能已经配置了 checkpoint，应以作业的实际配置为准。

为什么写入后的文件可能无法直接查看？

源表的 'format' = 'json' 用于解析 Kafka 消息，不决定 Iceberg 数据文件的格式。例如：

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

这条消息会被解析成 id INT 和 data STRING 两列，再交给 Iceberg writer 写入。

Iceberg 的 write.format.default 默认为 parquet。如果没有其他配置覆盖，数据文件通常是 Parquet。用文本编辑器打开这类文件时出现不可读字符，不能据此认定文件损坏。实际格式应检查表属性或元数据中的 file_format。Iceberg 写入配置

Iceberg 表目录还包含快照和 manifest 等元数据文件。判断写入结果是否正确，应查询 Iceberg 表，不要把 S3 目录中的任意对象当成原始 JSON 读取。

第一步：确认 checkpoint 和快照提交

对于 Iceberg 常见的 Flink streaming sink 实现，文件写入和表提交分为两个阶段，成功 checkpoint 后才会触发表提交。S3 中出现对象，并不等于这些对象已经进入可查询的 Iceberg 快照。Iceberg Flink 写入说明

先在 Flink UI 的 Checkpoints 页面确认：

  • checkpoint 是否启用。
  • 是否出现状态为 Completed 的 checkpoint。
  • checkpoint 完成后，Iceberg 是否产生了对应的新快照。

如果实际未启用 checkpoint，可以在提交写入作业之前设置：

SET 'execution.runtime-mode' = 'streaming';
SET 'execution.checkpointing.interval' = '30 s';

INSERT INTO glue_catalog.benchmark_db.orders_iceberg
SELECT id, data FROM kafka_source;

这里的 30 s 是排查示例，生产环境应结合写入延迟和文件数量调整。该设置只影响之后提交的作业。重新提交前先停止旧写入作业，避免两个作业同时消费并向同一张表追加。

如果 checkpoint 已完成但快照没有更新，继续检查 committer 日志和 Glue 提交操作。确认 S3 上传成功还不够。若所用 Iceberg 版本提供相关指标，还可以观察 committedDataFilesRecordCount 和 elapsedSecondsSinceLastSuccessfulCommit；具体指标是否可用取决于实际版本与 sink 实现。

第二步：通过 Iceberg 查询验证数据

在单独的 SQL 会话中配置相同的 glue_catalog，执行批量查询，读取查询开始时的表快照：

SET 'execution.runtime-mode' = 'batch';

SELECT id, data
FROM glue_catalog.benchmark_db.orders_iceberg
WHERE id = 1;

SELECT COUNT(*) AS row_count
FROM glue_catalog.benchmark_db.orders_iceberg;

如果测试记录已经成功提交且未被后续操作删除，第一条查询应能看到 id = 1、data = 'hello'。这只是预期结果，目前没有实际查询输出。

还可以检查快照和当前文件。Flink 中 Iceberg 元数据表使用 $ 后缀：Iceberg 元数据查询

SELECT snapshot_id, committed_at, operation, summary
FROM glue_catalog.benchmark_db.`orders_iceberg$snapshots`;

SELECT file_path, file_format, record_count
FROM glue_catalog.benchmark_db.`orders_iceberg$files`;

这些结果有助于区分以下情况：

  • S3 有对象，但尚无包含测试数据的已提交快照。
  • 表查询正确，只是直接查看二进制文件时误判为损坏。
  • 快照已提交，但查询出现具体的解析、权限或文件读取错误。

只有出现第三种情况，才需要根据实际报错继续排查文件、读取器、依赖或权限问题。

第三步：核实超过 950 的计数来自哪里

原始材料没有截图，也没有提供指标名称，因此无法直接解释这个数字。

Flink 的 numRecordsIn 和 numRecordsOut 分别统计某个 task/operator 收到和发出的记录，不等同于整张 Iceberg 表的最终行数。应先确定显示该数字的具体算子和指标。Flink 指标定义

如果数字出现在 Kafka source 或接收源数据的下游算子，优先检查 topic 的历史消息。示例配置为：

'scan.startup.mode' = 'earliest-offset'

对于不从已有状态恢复的新作业，这意味着从各分区当前仍保留的最早 offset 开始读取，而不是只读取最近发送的那一条。设置相同的 properties.group.id 也不会把这一启动模式改成 group-offsets。Flink Kafka 启动位置说明

还要核对是否重复提交过作业、是否从 checkpoint/savepoint 恢复，以及生产者是否实际重复发送。恢复作业时，源读取位置通常由恢复状态决定，不能只看建表参数。

如果数字出现在 Iceberg writer 后面的 committer 等算子上，应核对该算子的输入内容。某些 sink 实现会在内部传递文件写入结果或提交消息，不能直接将这类算子的计数与 Kafka 业务消息数比较。

验证单条消息时，比较容易控制的办法是使用一个新的空测试 topic 和新的测试表：先启动消费作业，再发送一次消息，等待 checkpoint 和快照提交后查询。若改用 latest-offset，也要先启动消费再发送；从已有状态恢复时，该设置未必会重新决定读取位置。

还需要确认的配置

ADD JAR 建议放在使用 connector 的建表和查询语句之前。如果 EMR 已经加载该 JAR，应核对实际加载的版本，避免重复引入依赖。

目前只给出了 Flink、Kafka 和 Kafka connector JAR 的信息，没有给出 EMR release、Iceberg runtime 或 AWS 相关依赖版本。现有信息无法判断是否存在版本不兼容，也不能把 Kafka 4.x 直接认定为根因。应核对 Iceberg runtime 是否支持 Flink 1.20，以及 EMR 自带依赖是否与手动添加的 JAR 冲突。

另外，CREATE TABLE IF NOT EXISTS 不会重建已经存在的目标表。多次测试时，应检查现有表的 schema、属性和历史数据，避免把旧数据算作本次写入的结果。

排查时先取得三个结果：具体算子的指标名称、成功 checkpoint 的记录，以及 Iceberg 表和元数据表的查询输出。这些信息有助于判断问题出在消费、提交还是读取阶段。

备注：内容仅供参考。