学习 Flink（十六）：Streaming Parquet File

更新至 Flink 1.8 版本

Flink 支持将流数据以文件的形式写入文件系统（HDFS、本地文件系统），支持 CSV、JSON 面向行的存储格式和 Parquet 面向列的存储格式。

应用场景：Flink 消费 Kafka 数据进行实时处理，并将结果以 Avro / Parquet 格式写入 HDFS。之后，遍可以使用 Spark 或 MPP 进行进一步分析。

由于流数据本身是无界的，所以，流数据将数据写入到分桶（bucket）中。默认使用基于时间的分桶策略。在分桶中，又根据滚动策略，将输出拆分为 part 文件。

Flink 提供了两个分桶策略，分桶策略实现了 org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner 接口：

• BasePathBucketAssigner ，不分桶，所有文件写到根目录；

• DateTimeBucketAssigner ，基于系统时间分桶。

Flink 提供了两个滚动策略，滚动策略实现了 org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy 接口：

• DefaultRollingPolicy 当超过最大桶大小（默认为 128 MB），或超过了滚动周期（默认为 60 秒），或未写入数据处于不活跃状态超时（默认为 60 秒）的时候，滚动文件；

• OnCheckpointRollingPolicy 当 checkpoint 的时候，滚动文件。

依赖

编辑 pom.xml 文件，添加依赖：

<dependency>  
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-parquet</artifactId>
    <version>1.8.0</version>
</dependency>  
<dependency>  
    <groupId>org.apache.parquet</groupId>
    <artifactId>parquet-avro</artifactId>
    <version>1.10.0</version>
</dependency>  
<dependency>  
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-filesystem_2.11</artifactId>
    <version>1.8.0</version>
</dependency>  
<dependency>  
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-shaded-hadoop2</artifactId>
    <version>1.8.0</version>
</dependency>

Sink

StreamingFileSink<Event> parquetFileSink = StreamingFileSink  
        .forBulkFormat(new Path("hdfs://RESOURCE_MANAGER/data/event"), ParquetAvroWriters.forReflectRecord(Event.class))
        .withBucketAssigner(new DateTimeBucketAssigner<>("'date='yyyy-MM-dd'/hour='HH"))
        .build();

eventStream.addSink(parquetFileSink);

注意：Bulk-encoding 格式仅支持 OnCheckpointRollingPolicy ，即在每次 checkpoint 时滚动 part 文件。

总结

截止目前，Flink 的 Streaming File Sink 仍存在不少问题，如：

• 不支持写入到 Hive

• 写入 HDFS 产生大量小文件

参考

• Apache Flink 1.8 Documentation: Streaming File Sink