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