学习 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 产生大量小文件

参考