Flink 读取Kafka写入Hive

官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/connectors/streamfile_sink.html

在流式处理系统中，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<String> input = ...;

final StreamingFileSink<String> sink = StreamingFileSink
	.forRowFormat(new Path(outputPath), new SimpleStringEncoder<>("UTF-8"))
	.build();

input.addSink(sink);

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

    StreamingFileSink<LogTest> 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能否保证一致性语义？