Flink TableAPI写入S3 Parquet文件时分区列缺失问题咨询

这并非Flink的已知问题，而是Filesystem连接器符合设计预期的默认行为：当你使用PARTITIONED BY指定分区列时，这些列的值会被编码到输出目录的路径中（而非写入Parquet文件的内容里），以此实现与Hive兼容的分区策略，优化后续查询的分区裁剪效率。

你可以检查S3上的输出路径，应该能看到类似这样的目录结构：

<S3 path>/record_id=abc123/source_name=user_service/date=2024-05-20/

分区列的信息通过目录层级来表达，不会出现在Parquet文件的Schema中。

额外注意点 #

你的INSERT语句存在一个潜在的易出错点：SELECT中用record_date作为别名，但sink表对应的列是date，当前是靠列位置匹配才生效。建议显式指定列名，避免后续表结构变更导致的字段映射错误：

INSERT INTO data_to_sink (record_id, request_id, source_name, event_type, event_name, `date`, results_count)
SELECT 
  record_id, 
  request_id, 
  source_name, 
  event_type, 
  event_name,
  DATE_FORMAT(TUMBLE_END(proc_time, INTERVAL '2' MINUTE), 'yyyy-MM-dd'), 
  COUNT(*)
FROM data_from_source
GROUP BY record_id, request_id, source_name, event_type, event_name, TUMBLE(proc_time, INTERVAL '2' MINUTE);

如果需要将分区列写入Parquet文件 #

若业务需求必须让分区列同时出现在文件内容中，可采用以下两种方案：

方案1：手动构造动态分区路径（不使用PARTITIONED BY）

将分区列作为普通字段保留，同时在path参数中用模板变量生成分区目录：

CREATE TABLE data_to_sink (
    record_id STRING NOT NULL,
    request_id STRING NOT NULL,
    source_name STRING NOT NULL,
    event_type STRING NOT NULL,
    event_name STRING NOT NULL,
    `date` STRING,
    results_count BIGINT
) WITH (
    'connector' = 'filesystem',
    'path' = '<S3 path>/record_id={record_id}/source_name={source_name}/date={date}',
    'format' = 'parquet'
);

此方案下所有字段（包括原分区列）都会写入Parquet文件，同时目录会按字段值自动分区。

方案2：重复输出分区列（兼顾分区目录与文件内容）

在sink表中新增专门的分区字段，同时在SELECT中重复输出原分区列的值：

CREATE TABLE data_to_sink (
    record_id STRING NOT NULL,
    request_id STRING NOT NULL,
    source_name STRING NOT NULL,
    event_type STRING NOT NULL,
    event_name STRING NOT NULL,
    `date` STRING,
    results_count BIGINT,
    part_record_id STRING,
    part_source_name STRING,
    part_date STRING
) PARTITIONED BY (part_record_id, part_source_name, part_date) WITH (
    'connector' = 'filesystem',
    'path' = '<S3 path>',
    'format' = 'parquet'
);

INSERT INTO data_to_sink
SELECT 
  record_id, 
  request_id, 
  source_name, 
  event_type, 
  event_name,
  DATE_FORMAT(TUMBLE_END(proc_time, INTERVAL '2' MINUTE), 'yyyy-MM-dd'), 
  COUNT(*),
  record_id, 
  source_name, 
  DATE_FORMAT(TUMBLE_END(proc_time, INTERVAL '2' MINUTE), 'yyyy-MM-dd')
FROM data_from_source
GROUP BY record_id, request_id, source_name, event_type, event_name, TUMBLE(proc_time, INTERVAL '2' MINUTE);

此方案会冗余存储分区列数据，需根据业务场景权衡使用。

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