说明

电商平台、游戏平台产生的用户数据，将数据放入 Kafka中的Topic，然后采用Flink去实时消费，并保存到HDFS上，保存后的数据可以构建数据仓库（如Hive）做数据分析，或是用于数据训练（算法模型）。

如下图所示：

File Sink

支持 流 和 批 模式下，flink将数据写入到文件系统中

官网

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/datastream/file_sink/

https://nightlies.apache.org/flink/flink-docs-release-1.12/dev/connectors/file_sink.html

桶

提示： 桶，相当于文件夹

File Sink 会将数据写入到桶中。由于输入流可能是无界的，因此每个桶中的数据被划分为 多个有限大小的文件。

如何分桶是可以配置的，默认使用 基于时间的分桶策略，这种策略每个小时创建一个新的桶，桶中包含的文件将记录所有该小时内从流中接收到的数据。

桶目录中的输出数据会被划分为 多个文件（part file）。

文件（part file）将根据 滚动策略创建，默认的策略是 根据文件大小 和 超时时间来滚动文件。

超时时间：指打开文件的最长持续时间，以及文件关闭前的最长非活动时间。

状态

In-progress

当前文件正在写入中
文件命名规则是：.part-<subtaskIndex>-<partFileIndex>.inprogress.<UUID>

提示：在 flink 1.12 本地测试时，文件命名规则是.part-<UUID>-<partFileIndex>.inprogress.<UUID>

注意： 由于是 . 开头，在linux、flink、hive 中，认为是隐藏文件，不会读取

Pending

当处于 In-progress 状态的文件关闭（closed）了，就变为 Pending 状态

Finished

在成功的 Checkpoint 后，Pending 状态将变为 Finished 状态

文件名命名规则：part-<subtaskIndex>-<partFileIndex>

提示：在 flink 1.12 本地测试时，文件命名规则是.part-<UUID>-<partFileIndex>

checkpoint

FileSink 在 STREAMING 模式的时候，必须 开启 checkpoint，否则每个文件一直处于 inprogress 或者 pending 状态

滚动策略

FileSink<Tuple2<Integer,Long>> sink = FileSink
    .<Tuple2<Integer,Long>> forRowFormat(
            new Path("./res") // 保存到本地。保存到hdfs的写法：hdfs://hadoop1:8020/flink_filesink
            , new SimpleStringEncoder<>("UTF-8"))
    .withRollingPolicy( // 指定滚动策略
            DefaultRollingPolicy.builder()
                    // 如果接收数据，每 15 分钟就滚动生成一个新文件
                    .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                    // 最近 5 分钟没有收到数据，就滚动生成一个新文件
                    .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                    // 文件大小达到 120 MB （写入最后一条记录后），就滚动生成一个新文件
                    .withMaxPartSize(120 * 1024 * 1024) // 每个文件最大 128M
                    .build())
    .build();

解释：

• 将数据分配给默认的一小时的时间桶，每个小时创建一个新的桶。

• 指定滚动策略：

  • 如果接收数据，每 15 分钟就滚动生成一个新文件
  • 最近 5 分钟没有收到数据，就滚动生成一个新文件
  • 文件大小达到 120MB （写入最后一条记录后），就滚动生成一个新文件，在 HDFS 上，防止大于 128MB ，切分文件

自定义

• 可自定义桶名（文件夹名）
• 可自定义文件的前缀、后缀

例子-本地测试

pom.xml

添加下面依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>

全部依赖

<dependencies>
    <!--java-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>

    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.10.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-files</artifactId>
        <version>${flink.version}</version>
    </dependency>

</dependencies>

java

package std;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;
import java.util.concurrent.TimeUnit;


public class Std9FileSink {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        checkpoint(env);
        // 转换操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso = env.addSource(buildKafka())
                .flatMap(new FlatMapFunImpl())
                .keyBy(value -> value.f0)
                .sum(1);

        FileSink<Tuple2<String, Integer>> sink = FileSink
                .<Tuple2<String, Integer>> forRowFormat(
                        new Path("./result/filesink") // 保存到本地。保存到hdfs的写法：hdfs://hadoop1:8020/flink_filesink
                        , new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy( // 指定滚动策略
                        DefaultRollingPolicy.builder()
                                // 如果接收数据，每 20秒 就滚动生成一个新文件
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(20))
                                // 最近 10秒 没有收到数据，就滚动生成一个新文件
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(10))
                                // 文件大小达到 120KB （写入最后一条记录后），就滚动生成一个新文件
                                .withMaxPartSize(120 )
                                .build())
                .build();

        soso2.sinkTo(sink);
        soso.print();
        env.execute("myjob");
    }

    private static FlinkKafkaConsumer<String> buildKafka(){
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        return fkc;
    }
    static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

            String[] arr = value.split(",");
            for(String item:arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }

    private static void checkpoint(StreamExecutionEnvironment env){
        // 必设置，每隔 5000 ms 执行一次checkpoint。
        env.enableCheckpointing(5000);//一般秒级、分钟级
        //设置保存文件路径，生产环境中应该是 HDFS
//        env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
        env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
        //env.setStateBackend(new RocksDBStateBackend("file:///D:/bigdata/flink_java/state/RocksDB",true));
        // 设置2个checkpoint之间最少等待时间，这里至少要等到500ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //checkpoint容忍的错误数量，达到此数量，就让该任务失败，默认值是：0。这里容忍3个错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //取消任务时，是否清理checkpoint
        // true：取消任务时，会删除外部checkpoint，默认值
        // false：取消任务时，保留外部checkpoint
        //这里设置不删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置checkpoint的执行模式：EXACTLY_ONCE，默认
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间，如果在60s内尚未完成说明本次checkpoint失败，则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);//这里设置10分钟
        // 设置同一时间有多少个checkpoint可以同时执行。与 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); 可能冲突
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//默认为1
        // 重启策略：程序出现异常时，重启2次，每次延迟3秒中重启，超过2次退出程序
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 3000));

    }
}

解释：

• 将数据分配给默认的一小时的时间桶，每个小时创建一个新的桶。

• 指定滚动策略：

  • 如果接收数据，每 15 分钟就滚动生成一个新文件
  • 最近 5 分钟没有收到数据，就滚动生成一个新文件
  • 文件大小达到 120 KB （写入最后一条记录后），就滚动生成一个新文件

测试

在kafka 反复发送下面数据：

李雷,韩梅梅,lucy
lili,lucy,李雷
韩梅梅,lucy

生成文件如下：

例子-HDFS

pom.xml

增加下面依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>
<!-- 可略 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.0.3</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.0.3</version>
</dependency>
<!-- 可略 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>3.0.3</version>
</dependency>

全部依赖

<dependencies>
    <!--java-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>

    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.10.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-files</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- 可略 -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>3.0.3</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.0.3</version>
    </dependency>
    <!-- 可略 -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>3.0.3</version>
    </dependency>

</dependencies>

java

package std;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;
import java.util.concurrent.TimeUnit;


public class StdFileSinkHDFS {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        checkpoint(env);
        // 转换操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso = env.addSource(buildKafka())
                .flatMap(new FlatMapFunImpl())
                .keyBy(value -> value.f0)
                .sum(1);

        FileSink<Tuple2<String, Integer>> sink = FileSink
                .<Tuple2<String, Integer>> forRowFormat(
                        new Path("hdfs://hadoop1:8020/flink_sink") // 保存到本地。保存到hdfs的写法：hdfs://hadoop1:8020/flink_filesink
                        , new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy( // 指定滚动策略
                        DefaultRollingPolicy.builder()
                                // 如果接收数据，每 20秒 就滚动生成一个新文件
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(20))
                                // 最近 10秒 没有收到数据，就滚动生成一个新文件
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(10))
                                // 文件大小达到 120KB （写入最后一条记录后），就滚动生成一个新文件
                                .withMaxPartSize(120 )
                                .build())
                .build();

        soso2.sinkTo(sink);
        soso.print();
        env.execute("myjob");
    }

    private static FlinkKafkaConsumer<String> buildKafka(){
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        return fkc;
    }
    static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

            String[] arr = value.split(",");
            for(String item:arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }

    private static void checkpoint(StreamExecutionEnvironment env){
        // 必设置，每隔 5000 ms 执行一次checkpoint。
        env.enableCheckpointing(5000);//一般秒级、分钟级
        //设置保存文件路径，生产环境中应该是 HDFS
//        env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
        env.setStateBackend(new FsStateBackend("hdfs://hadoop1:8020/flink_cp"));
        //env.setStateBackend(new RocksDBStateBackend("file:///D:/bigdata/flink_java/state/RocksDB",true));
        // 设置2个checkpoint之间最少等待时间，这里至少要等到500ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //checkpoint容忍的错误数量，达到此数量，就让该任务失败，默认值是：0。这里容忍3个错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //取消任务时，是否清理checkpoint
        // true：取消任务时，会删除外部checkpoint，默认值
        // false：取消任务时，保留外部checkpoint
        //这里设置不删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置checkpoint的执行模式：EXACTLY_ONCE，默认
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间，如果在60s内尚未完成说明本次checkpoint失败，则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);//这里设置10分钟
        // 设置同一时间有多少个checkpoint可以同时执行。与 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); 可能冲突
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//默认为1
        // 重启策略：程序出现异常时，重启2次，每次延迟3秒中重启，超过2次退出程序
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 3000));

    }
}

测试

在kafka 反复发送下面数据：

李雷,韩梅梅,lucy
lili,lucy,李雷
韩梅梅,lucy

参考：
https://www.cnblogs.com/smartloli/p/12499142.html

原文出处：/show_1IX44wEHhKEc.html