如何用Flink SQL读取含S3路径的Kafka事件？

仅通过Flink SQL可以实现两种情况的处理，但有版本和配置限制；如果遇到复杂场景，再考虑转为流处理调用S3 API。

一、仅用Flink SQL实现的方案（Flink 1.15+版本支持） #

核心思路是先判断contentJson的类型，再分别处理：

1. 用JSON_VALID函数判断字段是否为合法JSON，是则直接提取字段；
2. 若为S3路径，利用Flink内置的FILES表函数读取对应S3文件的内容，再提取JSON字段。

前提条件 #

• 确保Flink已配置AWS S3访问权限（在flink-conf.yaml中配置s3.access-key、s3.secret-key等参数）；
• S3路径需符合标准格式（如s3://bucket-name/folder/file.json，若原始路径是/bucket-name/...，需用CONCAT('s3://', contentJson)拼接成标准路径）。

示例SQL #

INSERT INTO final_table
SELECT
    JSON_VALUE(header, '$.some-path-json') AS value_1,
    CASE
        -- 处理contentJson为JSON字符串的情况
        WHEN JSON_VALID(contentJson) THEN JSON_VALUE(contentJson, '$.some-path-json')
        -- 处理contentJson为S3路径的情况：读取S3文件后提取字段
        ELSE JSON_VALUE(
            (SELECT CAST(content AS STRING) FROM FILES(
                PATH = CONCAT('s3://', contentJson), -- 拼接标准S3路径
                FORMAT = 'json'
            )),
            '$.some-path-json'
        )
    END AS value_2
FROM table_name;

注意事项 #

• 若大量事件为S3路径，频繁读取S3会带来性能开销，建议评估数据量和延迟要求；
• FILES函数仅支持Flink 1.15及以上版本，旧版本无法使用该方案。

二、需要转为流处理的场景 #

如果遇到以下情况，建议使用Flink DataStream API结合AWS SDK实现：

• 使用Flink 1.15以下版本，无FILES函数支持；
• 需要处理复杂S3交互逻辑（如临时凭证、文件不存在的异常重试、批量读取缓存等）；
• 对S3文件内容有自定义解析、转换需求，SQL内置函数无法满足。

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