官方文档：

在流式处理系统中，Flink和kafka的结合很是经典。我们可以通过Flink消费Kafka数据，层层处理后，丢到Kafka另一个Topic，下游再处理该Topic的数据。而对于OLAP查询需求，我们往往需要将数据输出到 Hive。一般的，我们使用Parquet格式来存储（Spark对parquet的支持较好）。

Flink提供了bucket sink的模式将流式数据写入到文件中，在官方给的demo中，代码如下

import org.apache.flink.api.common.serialization.SimpleStringEncoder;import org.apache.flink.core.fs.Path;import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;DataStream
    
      input = ...;final StreamingFileSink
     
       sink = StreamingFileSink	.forRowFormat(new Path(outputPath), new SimpleStringEncoder<>("UTF-8"))	.build();input.addSink(sink);
     
    

为了使用Parquet格式，我们还需要转换代码：

StreamingFileSink
    
      streamingFileSink = StreamingFileSink.        forBulkFormat(new Path(outputPath), ParquetAvroWriters.forReflectRecord(LogTest.class))        .withBucketAssigner(bucketAssigner)        .build();
    

在测试过程中，会发现目录创建了，但文件全为空且处于inprogress状态。经过多番搜索未解决该问题。最后在官方文档中发现了这么一句：

IMPORTANT: Bulk-encoding formats can only be combined with the `OnCheckpointRollingPolicy`, which rolls the in-progress part file on every checkpoint.

这说明Flink将一直缓存从Flink消费出来的数据，只有当Checkpoint 触发的时候，才把数据刷新到目标目录--即我们定义的parquet路径中。 加上启用CheckPoint之后，重新执行程序，可以发现文件成功写入了。

env.enableCheckpointing(3000);

其他思考：消费kafka输出到Parquet这一个过程，Flink能否保证一致性语义？