flink实战样例

flink概念介绍

flink源码以及样例等：https://github.com/apache/flink
flink各个版本开发文档：https://nightlies.apache.org/flink/

这StreamExecutionEnvironment是所有 Flink 程序的基础（创建批处理请使用ExecutionEnvironment）。创建一个执行环境,表示当前执行程序的上下文,类似于SparkContext.

1：StreamExecutionEnvironment的API解释

 protected static void envSetConfig(  LocalStreamEnvironment env) {
        env.setBufferTimeout(1000);
        env.setMaxParallelism(10);//设置最大并行度
        env.setParallelism(8);//设置并行度

        //设置重启策略
        env.enableCheckpointing(5000);//设置重启策略必须先开启enableCheckpointing
        env.setRestartStrategy(new RestartStrategies.FailureRateRestartStrategyConfiguration(5, Time.milliseconds(100),Time.milliseconds(100)));

        //设置状态后端的存储位置。包括内存，文件系统等
        //1:内存
        env.setStateBackend(new MemoryStateBackend());
        //2:文件系统时必须指定checkpoint存储路径
        env.setStateBackend(new FsStateBackend("C:\Users\Administrator.SC-201905261418\Desktop\testData\flink"));

        //设置时间特性，用于窗口和水印使用
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    }

为了保证task任务或者算子执行过程中的失败能够恢复，启用检查点存储算子的执行状态（快照方式）。失败时从最新的快照进行恢复。
相关的配置参数如下：

env.enableCheckpointing(1000);
        // advanced options:
        // set mode to exactly-once (this is the default)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // 设置一次checkpoint完成的最长时间，超过取消此次checkpoint
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //只允许一个检查点同时进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 启用作业取消后保留的外部化检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

在Java/Scala API上可以通过 stream.keyBy(…) 得到 KeyedStream，在Python API上可以通过 stream.key_by(…) 得到 KeyedStream。

所有支持的状态类型如下所示：

ValueState: 保存一个可以更新和检索的值（如上所述，每个值都对应到当前的输入数据的 key，因此算子接收到的每个 key 都可能对应一个值）。 这个值可以通过 update(T) 进行更新，通过 T value() 进行检索。

ListState: 保存一个元素的列表。可以往这个列表中追加数据，并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List) 进行添加元素，通过 Iterable get() 获得整个列表。还可以通过 update(List) 覆盖当前的列表。

ReducingState: 保存一个单值，表示添加到状态的所有值的聚合。接口与 ListState 类似，但使用 add(T) 增加元素，会使用提供的 ReduceFunction 进行聚合。

AggregatingState: 保留一个单值，表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与 添加到状态的元素的类型不同。 接口与 ListState 类似，但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合。

MapState: 维护了一个映射列表。 你可以添加键值对到状态中，也可以获得反映当前所有映射的迭代器。使用 put(UK，UV) 或者 putAll(Map) 添加映射。 使用 get(UK) 检索特定 key。 使用 entries()，keys() 和 values() 分别检索映射、键和值的可迭代视图。你还可以通过 isEmpty() 来判断是否包含任何键值对。

状态必须通过RichFunction函数才可以创建，创建StateDescriptor，才能得到对应的状态句柄。 这保存了状态名称、状态所持有值的类型。

使用示例：

  stream.keyBy(s -> s).flatMap(new RichFlatMapFunction() {
            //状态作为实例变量进行定义
            private ValueState vlaueSateTest;

            @Override //open方法只执行一次
            public void open(Configuration parameters) throws Exception {
                //声明状态：StateDescriptor 包含状态名称和有关状态所存值的类型

                //1：声明valuestate
                ValueStateDescriptor stateDescriptor = new ValueStateDescriptor<>(
                        "vlaueSateTest", //状态的名字。必须保证唯一性，后续通过name获取
                        Integer.class); //存储值的类型。
                //可以设置状态的过期时间等。
                //StateTtlConfig用于配置状态相关参数
                StateTtlConfig build = StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.minutes(1)).cleanupFullSnapshot().build();

                stateDescriptor.enableTimeToLive(build);

                //获取状态必须从getRuntimeContext进行get
                this.vlaueSateTest = getRuntimeContext().getState(stateDescriptor);
                //2:声明
                ListState listStateTest = getRuntimeContext().getListState(new ListStateDescriptor("listStateTest", ValueTypeInfo.INT_VALUE_TYPE_INFO));
            }

            @Override
            public void flatMap(Integer integer, Collector