【技术分享】Flink应用开发入门

1StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//禁止输出job的状态更新日志,默认是会输出的
2env.getConfig().disableSysoutLogging();//job失败后使用延迟重启策略,延迟10秒重启job,最多重启4次
3env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
4//启用checkPoint,每5秒执行一次CheckPoint 注:当遇到故障时，flink会停止分布式数据流。系统会重启所有的operator,重至到最近成功的checkpoint。输入重置相应的状态快照位置，来保证重置位置的正确性。
5env.enableCheckpointing(5000); 

1SourceFunction<String> source = ...；//构建自己的数据源。
2DataStream<String> input = env.addSource(source)；//添加数据源到Flink执行环境中。

 1DataStream<Map<String, String>> baseDataStream = input
 2    .flatMap(new FlatMapFunction<String, Map<String, String>>() {
 3
 4 private static final long serialVersionUID = 1L;
 5
 6 @Override
 7     public void flatMap(String value, Collector<Map<String, String>> out) {
 8
 9      /**
10       * 处理数据转换
11       */
12      Map<String, String> data = ... ;//将source流出的value数据转成map;
13      if (data != null) {
14       out.collect(data);//收集转换后的数据
15      }
16     }
17    });

 1//在这里我们生成一个业务系统中比较常用的窗口，基于时间的滑动窗口。
 2WindowedStream<ReduceBean, Tuple, TimeWindow> windowedStream = null;
 3windowedStream = outputDataStream.keyBy("keyField")  //使用指定的字段对窗口进行分组，keyField为ReduceBean中的字段名称，当然也可以使用下标位置(int值)来指定分组字段。
 4.timeWindow(Time.seconds(60) , Time.seconds(10));  //使用滑动时间窗口,窗口大小为60秒,窗口中每10秒一个窗格(每10秒计算最近60秒的数据)。
 5//使用reduce function处理自定义汇集计算
 6SingleOutputStreamOperator<ReduceBean> outputStream = null;
 7outputStream = windowedStream.reduce(new ReduceFunction<ReduceBean>() {
 8@Override
 9public ReduceBean reduce(ReduceBean a, ReduceBean b) {
10return new ReduceBean(a.keyField, a.value+b.value);  // reduce函数中实现求和计算,根据自己的业务需求可以自己实现这里的reduce处理逻辑,返回一个计算后的ReduceBean即可(ReduceBean为自己定义的一个序列化类)
11        }
12       });

1//构建好自定义sinkFunction后,追加到Flink的执行计划中
2outputStream.addSink(sinkFunction);
3//最后启动执行job即可
4env.execute(jobName);