Flink（十一）Flink连接Kafka输出到HDFS

电玩女神 2022-01-26 16:27 432阅读 0赞

## 一、配置项目的依赖 ##

> 其中flink-connector-filesystem\_2.11是将Hadoop作为Flink的BucketingSink接入，
> 
> hadoop-hdfs、hadoop-common、hadoop-client解决Jar包依赖的问题，2.7.3为hadoop的版本号。

<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状态）**时，将刷新并关闭打开的部件文件。如果存储桶最近未写入，则视为非活动状态。默认情况下，**接收器每分钟检查一次非活动存储桶，并关闭任何超过一分钟未写入的存储桶**。

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();

> 存储桶文件的格式

/base/path/{date-time}/part-{parallel-task}-{count}

`date-time`我们从日期/时间格式获取的字符串，`parallel-task`是并行接收器实例的索引，`count`是由于批处理大小创建的块文件的运行数。