Apache Flink 超简朴讲解，Java中的分布式流盘算框架

嘿，各人好！今天咱们聊聊 Java 生态中一个超酷炫的工具——Apache Flink。它就像一把瑞士军刀，专门用来处置惩罚那些海量的、及时活动的数据流。想象一下，如果你有一条永不停息的数据河道，而你想要在这条河里捞出有效的信息，Flink 就是你的好帮忙！
戳底部手刺，一起变现

一、为啥有 Flink？

在互联网时代，数据像潮水一样涌来，企业不仅必要存储这些数据，更希望可以或许迅速从中发掘出价值。传统的批处置惩罚方法有点跟不上节奏了，因为它们处置惩罚完一批数据得等下一批，这就导致了延迟。于是，一些早期的流处置惩罚框架出现了，比如 Storm 和 Spark Streaming。但是这些框架也有些小问题，比如状态管理不敷机动，大概在保证数据一致性方面做得不是特别好。
Flink 就是在这样的背景下诞生的。它的开辟者们想：“如果能有个东西既能做到及时处置惩罚，又能确保数据不丢不乱，那该多好啊！”所以他们造出了 Flink。这个家伙从一开始就被计划成可以处置惩罚无穷的数据流，并且可以或许保证即使在网络故障或节点瓦解的环境下也能保持数据的一致性和准确性。
二、Flink 都干啥用？

说白了，Flink 可以用来做许多风趣的变乱：

• 及时数据分析：比如说你想知道现在有多少人在看你的直播，大概哪个产物最受欢迎，Flink 可以帮你及时统计。
  
• 物联网 (IoT) 数据处置惩罚：想象一下智能家居系统里的温度传感器、湿度传感器，Flink 可以及时收集和分析这些数据，告诉你家里是不是太热了大概太湿了。
  
• 金融风控：银行和支付平台可以用 Flink 来检测异常生意业务，防止诈骗。
  
• 推荐系统：电商网站可以根据用户的欣赏历史给出个性化的商品推荐，让购物体验更贴心。
  
• A/B 测试与实验：测试差别的广告计谋效果怎样，看看哪种方式最吸引人。
  

三、Flink 的基础知识

1. DataStream API

DataStream API 是 Flink 的核心接口之一，就比如是搭建乐高积木时使用的各种形状的小块。你可以用它来做许多变乱，比如把数据映射成新的情势、过滤掉不必要的东西、按某个字段分组等等。下面是一个简朴的例子，我们读取一段文本，然后盘算每个单词出现的次数。

1. import org.apache.flink.api.common.functions.FlatMapFunction;
2. import org.apache.flink.api.java.tuple.Tuple2;
3. import org.apache.flink.streaming.api.datastream.DataStream;
4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. import org.apache.flink.util.Collector;
7. public class SimpleWordCount {
8.     public static void main(String[] args) throws Exception {
9.         // 创建执行环境
10.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
12.         // 假设有一个文本文件作为输入源
13.         DataStream<String> text = env.fromElements("hello world", "hello flink");
15.         // 解析数据，按空格分割字符串，并映射为 (word, 1) 的键值对
16.         DataStream<Tuple2<String, Integer>> counts =
17.             text
18.                 .flatMap(new Tokenizer())
19.                 .keyBy(value -> value.f0) // 按第一个字段分组
20.                 .sum(1);                  // 对第二个字段求和
22.         // 输出结果到控制台
23.         counts.print();
25.         // 执行程序
26.         env.execute("Simple Word Count");
27.     }
29.     // 自定义 FlatMapFunction 实现，用于将每行文本拆分为多个单词
30.     public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
32.         @Override
33.         public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
34.             // 将每一行按照空格分割成单词
35.             String[] tokens = value.toLowerCase().split("\\W+");
37.             // 发射每个单词作为一个元组 (word, 1)
38.             for (String token : tokens) {
39.                 if (token.length() > 0) {
40.                     out.collect(new Tuple2<>(token, 1));
41.                 }
42.             }
43.         }
44.     }
45. }

复制代码

戳底部手刺，一起变现

2. Windowing

有时候我们必要对一段时间内的数据举行聚合操作，比如盘算过去一分钟内发生了多少次点击。这就是窗口函数的作用。Flink 支持多种范例的窗口，比如滚动窗口（固定时间间隔）、滑动窗口（重叠的时间段）和会话窗口（基于活动间隙）。下面的例子展示了怎样使用滚动窗口来盘算每个单词在过去5秒内出现的次数。

1. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
2. import org.apache.flink.api.common.functions.AggregateFunction;
3. import org.apache.flink.api.java.tuple.Tuple2;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
6. import org.apache.flink.streaming.api.windowing.time.Time;
7. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
8. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
9. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
10. import org.apache.flink.util.Collector;
12. public class WordCountWithWindow {
13.     public static void main(String[] args) throws Exception {
14.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.         // 模拟一个持续不断的文本流
17.         DataStream<String> text = env.fromElements("hello world", "hello flink", "hello again");
19.         // 设置事件时间戳和水印策略（这里简化处理）
20.         DataStream<Tuple2<String, Integer>> wordCounts = text
21.             .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
22.                 for (String word : line.split(" ")) {
23.                     if (!word.isEmpty()) {
24.                         out.collect(Tuple2.of(word, 1));
25.                     }
26.                 }
27.             })
28.             .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Integer>>forMonotonousTimestamps()
29.                 .withTimestampAssigner((event, timestamp) -> System.currentTimeMillis()))
30.             .keyBy(value -> value.f0) // 按单词分组
31.             .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 定义5秒的滚动窗口
32.             .aggregate(new WordCountAggregate(), new ResultFormatter());
34.         wordCounts.print();
35.         env.execute("Word Count with Windows");
36.     }
38.     // 自定义聚合函数
39.     public static class WordCountAggregate implements AggregateFunction<Tuple2<String, Integer>, Integer, Tuple2<String, Integer>> {
40.         @Override
41.         public Integer createAccumulator() {
42.             return 0;
43.         }
45.         @Override
46.         public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
47.             return accumulator + value.f1;
48.         }
50.         @Override
51.         public Tuple2<String, Integer> getResult(Integer accumulator) {
52.             return Tuple2.of("count", accumulator);
53.         }
55.         @Override
56.         public Integer merge(Integer a, Integer b) {
57.             return a + b;
58.         }
59.     }
61.     // 格式化输出结果
62.     public static class ResultFormatter implements WindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> {
63.         @Override
64.         public void apply(String key, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) {
65.             int sum = 0;
66.             for (Tuple2<String, Integer> in : input) {
67.                 sum += in.f1;
68.             }
69.             out.collect(Tuple2.of(key, sum));
70.         }
71.     }
72. }

复制代码

3. State Management

Flink 还允许你在运行过程中保存临时的状态信息，这就像玩游戏时存档一样。比如说，如果你正在统计某个用户的行为模式，那么你可以把已经看到的数据暂时存起来，比及后面再继续处置惩罚。下面是使用 ValueState 来跟踪每个单词的最新计数。

1. import org.apache.flink.api.common.state.ValueState;
2. import org.apache.flink.api.common.state.ValueStateDescriptor;
3. import org.apache.flink.api.common.functions.RichFlatMapFunction;
4. import org.apache.flink.api.java.tuple.Tuple2;
5. import org.apache.flink.streaming.api.datastream.DataStream;
6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7. import org.apache.flink.util.Collector;
9. public class WordCountWithState {
10.     public static void main(String[] args) throws Exception {
11.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
13.         // 模拟一个持续不断的文本流
14.         DataStream<String> text = env.fromElements("hello world", "hello flink", "hello again");
16.         // 使用 RichFlatMapFunction 来访问状态
17.         DataStream<Tuple2<String, Integer>> wordCounts = text
18.             .flatMap(new Tokenizer())
19.             .keyBy(value -> value.f0) // 按单词分组
20.             .flatMap(new CountWithState());
22.         wordCounts.print();
23.         env.execute("Word Count with State");
24.     }
26.     // 自定义 FlatMapFunction 实现，用于将每行文本拆分为多个单词
27.     public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
28.         @Override
29.         public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
30.             for (String word : value.split(" ")) {
31.                 if (!word.isEmpty()) {
32.                     out.collect(Tuple2.of(word, 1));
33.                 }
34.             }
35.         }
36.     }
38.     // 使用 ValueState 来跟踪每个单词的计数
39.     public static final class CountWithState extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {
40.         private transient ValueState<Integer> countState;
42.         @Override
43.         public void open(Configuration parameters) {
44.             ValueStateDescriptor<Integer> descriptor =
45.                 new ValueStateDescriptor<>("word-count", Integer.class);
46.             countState = getRuntimeContext().getState(descriptor);
47.         }
49.         @Override
50.         public void flatMap(Tuple2<String, Integer> input, Collector<Tuple2<String, Integer>> out) throws Exception {
51.             Integer currentCount = countState.value();
52.             if (currentCount == null) {
53.                 currentCount = 0;
54.             }
55.             currentCount += input.f1;
56.             countState.update(currentCount);
57.             out.collect(new Tuple2<>(input.f0, currentCount));
58.         }
59.     }
60. }

复制代码

4. Checkpointing

为了防止不测发生，Flink 提供了一个叫做 checkpoint 的功能，就像是给步伐拍快照一样。每隔一段时间，它会把当前的状态保存下来，万一之后出错了，就可以直接从近来的一个 checkpoint 开始重新运行，而不会丢失任何数据。默认环境下，Flink 会主动为你设置 checkpoint，但你也可以自定义设置，比如多久拍一次快照，大概把快照存在哪里。

1. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
2. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
4. public class EnableCheckpointing {
5.     public static void main(String[] args) throws Exception {
6.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
8.         // 启用 checkpoint 功能，每隔5分钟创建一次检查点
9.         env.enableCheckpointing(5 * 60 * 1000);
11.         // 设置 checkpoint 存储位置
12.         env.setStateBackend(new FsStateBackend("file:///path/to/checkpoints"));
14.         // ... 其他代码 ...
15.         
16.         env.execute("Enable Checkpointing Example");
17.     }
18. }

复制代码

5. Fault Tolerance

最后，Flink 还内置了许多容错机制，确保即使部分组件挂掉了也不会影响整个系统的正常工作。比方，它支持使命重试、失败隔离等功能，这些都是为了让我们的应用更加稳定可靠。
戳底部手刺，一起变现


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。