Apache Flink按日期分区写HDFS Parquet及空文件问题求助

嘿，针对你用Flink 1.4.0 + HDFS 2.7.3遇到的两个问题，我给你整理了贴合版本特性的具体解决方案：

在Flink 1.4.0中，主要依赖BucketingSink实现分区写入，步骤如下：

1. 确定分区依据与格式 #

可以选择事件时间（从JSON数据中提取的业务日期）或处理时间作为分区键，最终生成类似hdfs://your-base-path/dt=2024-05-20/的Hive兼容分区目录。

2. 配置BucketingSink核心参数 #

• 先将Kafka摄入的JSON数据转换为Avro对象（Parquet与Avro结合的格式能被Hive直接识别）；
• 自定义分区策略，指定日期格式的分区目录；
• 使用官方提供的AvroParquetWriterFactory来生成符合规范的Parquet文件。

代码示例 #

// 1. 从Kafka读取JSON并转换为Avro GenericRecord
DataStream<GenericRecord> jsonStream = env.addSource(new FlinkKafkaConsumer010<>(
        "your-kafka-topic",
        new JSONKeyValueDeserializationSchema(false),
        kafkaConfig
))
.map(jsonObj -> {
    // 加载提前定义好的Avro Schema
    Schema avroSchema = new Schema.Parser().parse(new File("your-schema.avsc"));
    GenericRecord record = new GenericData.Record(avroSchema);
    // 解析JSON字段到Avro对象，提取日期作为分区键
    record.put("id", jsonObj.get("id"));
    record.put("dt", jsonObj.get("event_time").toString().substring(0, 10));
    // 其他字段映射...
    return record;
});

// 2. 配置BucketingSink
BucketingSink<GenericRecord> parquetSink = new BucketingSink<>("hdfs://your-base-path");
// 设置分区策略：按dt字段生成dt=yyyy-MM-dd格式的目录
parquetSink.setBucketer(new Bucketer<GenericRecord>() {
    @Override
    public Path getBucketPath(Clock clock, Path basePath, GenericRecord element) {
        String datePartition = element.get("dt").toString();
        return new Path(basePath + "/dt=" + datePartition);
    }
});
// 使用AvroParquetWriterFactory写入Parquet文件
parquetSink.setWriter(new AvroParquetWriterFactory<>(avroSchema));
// 配置滚动策略：避免小文件，15分钟或文件达1GB时滚动
parquetSink.setRollingPolicy(new DefaultRollingPolicy.Builder()
        .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
        .withMaxPartSize(1024 * 1024 * 1024) // 1GB
        .build());
// 设置临时文件前缀后缀，避免Hive读取未完成的文件
parquetSink.setPendingPrefix(".");
parquetSink.setPendingSuffix(".tmp");

// 3. 将数据写入HDFS
jsonStream.addSink(parquetSink);

这个问题在Flink 1.4.0中属于常见坑，核心原因和解决办法如下：

1. 问题根源 #

• 自定义Writer未正确处理flush和close逻辑，导致数据未实际落盘；
• BucketingSink的滚动策略与Checkpoint机制冲突，触发空文件生成；
• 自定义Writer未实现序列化，分布式环境下实例化异常导致无数据写入。

2. 具体解决办法 #

（1）优先使用官方AvroParquetWriterFactory

官方Factory已经封装了正确的写入、flush和close逻辑，避免手动实现出错。如果你之前是自定义ParquetWriter，直接替换为Factory即可。

（2）调整滚动策略与Checkpoint配置

开启Checkpoint并匹配滚动策略，避免无数据时生成空文件：

// 开启Checkpoint，5分钟一次，Exactly-Once语义
env.enableCheckpointing(TimeUnit.MINUTES.toMillis(5));
env.getCheckpointConfig().setCheckpointMode(CheckpointMode.EXACTLY_ONCE);
// 增加非活动超时，5分钟无数据就滚动文件（避免空文件一直挂着）
parquetSink.setRollingPolicy(new DefaultRollingPolicy.Builder()
        .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
        .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
        .withMaxPartSize(1024 * 1024 * 1024)
        .build());

（3）修复自定义Writer的生命周期逻辑

如果必须自定义Writer，一定要重写flush()和close()方法，同时实现序列化：

public class CustomAvroParquetWriter extends Writer<GenericRecord> {
    private transient AvroParquetWriter<GenericRecord> writer;
    private Path filePath;
    private Schema avroSchema;

    public CustomAvroParquetWriter(Path filePath, Schema avroSchema) {
        this.filePath = filePath;
        this.avroSchema = avroSchema;
        try {
            writer = new AvroParquetWriter<>(filePath, avroSchema);
        } catch (IOException e) {
            throw new RuntimeException("初始化ParquetWriter失败", e);
        }
    }

    @Override
    public void write(GenericRecord element) throws IOException {
        writer.write(element);
    }

    @Override
    public void flush() throws IOException {
        if (writer != null) {
            writer.flush();
        }
    }

    @Override
    public void close() throws IOException {
        if (writer != null) {
            writer.close();
        }
    }

    // 实现序列化与反序列化，适配分布式环境
    private void writeObject(ObjectOutputStream out) throws IOException {
        out.defaultWriteObject();
    }

    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        writer = new AvroParquetWriter<>(filePath, avroSchema);
    }
}

（4）检查HDFS配置

确保HDFS的dfs.client.block.write.replace-datanode-on-failure.policy设置为NEVER，避免写入时因DataNode问题产生空文件；同时确认Flink任务有目标路径的写入权限。

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