一段Flink连接Kafka输出到HDFS的代码

大数据真好玩

发布于 2020-06-12 10:28:24

1.6K0

一、配置项目的依赖

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

存储桶文件的格式

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

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-06-05，如有侵权请联系 cloudcommunity@tencent.com 删除

hadoop