Flink开发入门简单案例
- 0.简介
- 1.订单数据生成器
- 1.1 新建工程TestFlink
- 1.2 在pom.xml中引入Flink依赖包
- 1.3 订单数据生成类
- +订单类（Item）
- +订单生成数据流类
- +测试订单生成类
- 2.订单统计
- 2.1 仅统计订单中商品的件数
- 2.2 同时统计商品数量和金额
0.简介
本案例通过一个简单的订单数据生成器生成随机订单数据，基于这个生成器运用Flink DataStream API开发程序统计订单数量和金额，模拟交易看板数据。
环境：IDEA作为IDE，Flink 1.13.2，Scala 2.12，Java 1.8
1.订单数据生成器
利用 Flink 提供的自定义 Source 来实现一个自定义的实时数据生成器
1.1 新建工程TestFlink
在IDEA中新建工程（new Project），选择“Maven Archetype”，catalog选最基础的类型，如下图所示：
1.2 在pom.xml中引入Flink依赖包
分别引入flink-java, flink-streaming-java, flink-client三个包，配置如下：
<properties>
<flink.version>1.13.2</flink.version>
<scala.binary.version>2.12</scala.binary.version>
<slf4j.version>1.7.30</slf4j.version>
<mysql-connector.version>8.0.23</mysql-connector.version>
</properties>
<dependencies>
<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_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
上面的版本均采用变量定义，可以根据实际情况进行替换。
1.3 订单数据生成类
主要用到三个类，订单类，生成订单数据流类，测试类
+订单类（Item）
普通Java类，代码如下：
public class Item {
private String name;
private Integer id;
Item() {}
public String getName() {
return name;
}
void setName(String name) {
this.name = name;
}
private Integer getId() {
return this.id;
}
void setId(int id) {
this.id = id;
}
public String toString() {
return "Item { " +
"name='" + name + "\'" +
", id=" + id + "}";
}
}
+订单生成数据流类
用于生成订单类的数据流，代码如下：
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;
public class MyStreamingSource implements SourceFunction<Item> {
private boolean isRunning = true;
@Override
public void run(SourceContext<Item> sourceContext) throws Exception {
while(isRunning) {
Item item = generateItem();
sourceContext.collect(item);
Thread.sleep(1000);
}
}
private Item generateItem() {
int i = new Random().nextInt(1000);
Item item = new Item();
item.setId(i);
item.setCount((i % 4) + 1);
item.setSum(item.getCount() * (new Random().nextFloat() * 50));
return item;
}
@Override
public void cancel() {
isRunning = false;
}
}
+测试订单生成类
用于测试上述生成类，也是程序的主函数，代码如下
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class StreamingDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Item> text = env.addSource(new MyStreamingSource()).setParallelism(1);
DataStream<Item> item = text.map((MapFunction<Item, Item>) value-> value);
item.print().setParallelism(1);
String jobName = "user defined streaming source";
env.execute(jobName);
}
}
运行StreamingDemo的main函数，可以看到如下的输出结果：
2.订单统计
接下来，编写Flink算子，以窗口的方式统计一段时间内（5秒为例）订单的数量。
2.1 仅统计订单中商品的件数
思路如下：先读入订单数据源，将读到的订单Item对象映射为一个三元组Tuple3，再按5秒钟的窗口对订单中商品数量（count）进行求和（sum），核心代码如下：
DataStreamSource<Item> order = env.addSource(new MyStreamingSource()).setParallelism(1);
DataStream<Tuple3<Integer, Integer, Float>> mappedStream =
order.map(new MapFunction<Item, Tuple3<Integer, Integer, Float>>() {
@Override
public Tuple3<Integer, Integer, Float> map(Item item) throws Exception {
return new Tuple3<>(item.getId(), item.getCount(), item.getSum());
}
});
//定义窗口，每5秒一个
DataStream<Tuple3<Integer, Integer, Float>> windowedStream = mappedStream
.keyBy((item)->0) // 此处的分组方式未使用任何分组，所以使用了一个常量0
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum(1); // 统计三元组中序号为1（即count属性）的和
windowedStream.print();
运行后的结果如下：
可以看到，控制台中的输出，每一条数据前是5条数据，也就是5秒钟的窗口收集到5条（每秒产生1条），第2列为商品数量的和，实现了累加。
2.2 同时统计商品数量和金额
上述示例中，对windowedStream使用了sum进行统计求和，这种方式只能对一个字段进行，如果需要同时统计数量和金额，就必须采用另外一种方式，reduce进行统计。
reduce方法也是windowedStream提供的方法，代码如下：
DataStream<Tuple3<Integer, Integer, Float>> windowedStream = mappedStream
.keyBy((item)->0)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.reduce(new ReduceFunction<Tuple3<Integer, Integer, Float>>() {
@Override
public Tuple3<Integer, Integer, Float> reduce(Tuple3<Integer, Integer, Float> t1, Tuple3<Integer, Integer, Float> t2) throws Exception {
return new Tuple3<>(0, t1.f1 + t2.f1, t1.f2 + t2.f2);
}
});
这样，即可实现同时统计数量和金额，运行后效果如下：
可以看到，第2列是count的和，第3列也是sum的和，实现了同时统计。（第1列是在程序中直接映射为了0，因为id字段统计无意义）
本文章的部分案例借鉴自以下文章：Flink 常用的 DataSet 和 DataStream API，有需要的可以自行查看。
