一、配置项目的依赖
其中flink-connector-filesystem_2.11是将Hadoop作为Flink的BucketingSink接入，
hadoop-hdfs、hadoop-common、hadoop-client解决Jar包依赖的问题，2.7.3为hadoop的版本号。
代码语言：javascript复制<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem_2.11</artifactId>
<version>1.8.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.3</version>
</dependency>二、Flink启动程序
当数据到达时，分段接收器将按当前系统时间拆分，并使用日期时间模式"yyyy-MM-dd-HH"命名存储区。这种模式传递给DateTimeFormatter使用当前系统时间和东八时区（上海）来形成存储桶路径。每当遇到新日期时，都会创建一个新存储桶。每个存储桶本身都是一个包含多个块文件的目录：接收器的每个并行实例将创建自己的块文件，当块文件超过100MB或超过20分钟时，接收器也会创建新的块文件。当存储桶变为非活动状态（非in-progress状态）时，将刷新并关闭打开的部件文件。如果存储桶最近未写入，则视为非活动状态。默认情况下，接收器每分钟检查一次非活动存储桶，并关闭任何超过一分钟未写入的存储桶。
代码语言：javascript复制 StreamExecutionEnvironment flinkEnv = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取Kafka配置
Properties props = new Properties();
props.putAll(kafkaProperties.buildConsumerProperties());
// 创建Kafka-Source
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(KafkaTopic.TRACK_LOGS, new SimpleStringSchema(), props);
// 添加Kafka-Source
DataStreamSource<String> source = flinkEnv.addSource(consumer);
// 方式1：将数据导入Hadoop的文件夹
//recordData.writeAsText("hdfs://hadoop:9000/flink/");
// 方式2：将数据导入Hadoop的文件夹
BucketingSink<String> hadoopSink = new BucketingSink<>("hdfs://hadoop:9000/flink/");
// 使用东八区时间格式"yyyy-MM-dd--HH"命名存储区
hadoopSink.setBucketer(new DateTimeBucketer<>("yyyy-MM-dd--HH", ZoneId.of("Asia/Shanghai")));
// 下述两种条件满足其一时，创建新的块文件
// 条件1.设置块大小为100MB
hadoopSink.setBatchSize(1024 * 1024 * 100);
// 条件2.设置时间间隔20min
hadoopSink.setBatchRolloverInterval(20 * 60 * 1000);
// 设置块文件前缀
hadoopSink.setPendingPrefix("");
// 设置块文件后缀
hadoopSink.setPendingSuffix("");
// 设置运行中的文件前缀
hadoopSink.setInProgressPrefix(".");
// 添加Hadoop-Sink,处理相应逻辑
source.addSink(hadoopSink);
flinkEnv.execute();存储桶文件的格式
date-time我们从日期/时间格式获取的字符串，parallel-task是并行接收器实例的索引，count是由于批处理大小创建的块文件的运行数。
