Flink流join案例详解【含代码】
问题导读：
1、怎样关联流数据？
2、在Flink中，流Join有哪几种？
3、怎样利用window join关联商品与订单？
1 JOIN
自从Stream pipeline解决方案地成熟，流操作和关系型表结构操作的差距越来越小了。我们通过Flink这样的框架，可以进行高吞吐量的数据流执行非常密集的数据处理，例如：join、filter、aggregation。所以，接下来我们就来看看Flink的Stream join。
在介绍Stream join之前，我们先来回顾一下关系型数据的Join、以及数仓维度建模的Join。
1 RDBMS Join
在学习关系型数据库时，就对关系型数据库的JOIN非常熟悉。RDBMS中，一般有很多表，而为了满足范式设计，我们会将数据分开在不同的表中存储。而我们获取数据的时候，需要将这些表的数据集中起来，这时候，我们就要使用JOIN了。
MySQL支持多种JOIN，例如：LEFT JOIN、RIGHT JOIN、INNER JOIN。而每一种JOIN具备不同特点。
2 数仓维度建模Join
在数据仓库设计中，我们一般会采用维度建模方式，而事实表中有大量的维表外键ID以及大量指标。通过维度表外键ID，我们可以和维度表进行关联，然后通过不同维度来统计计算指标。
3 流Join
在实时处理中，事实数据（Event）源源不断地流入到Kafka中，这些事实数据可能是来自于网页中的用户访问日志、也有可能是MySQL中进行操作的binlog日志，但基本上可以确认的是，在这个事件流中，不是所有的数据都是完整的。绝大多数场景，原始日志数据需要和外部存储系统中的一些数据进行关联之后，才能得到更完整的数据。
为了进行数据的实时处理，我们需要在流处理系统将数据进行关联处理，才能得到最完整的数据。我们怎么样关联呢？假设，我们需要在Flink中进行实时指标统计，也就是我们需要用事件流中的数据与存储在MySQL中的维表在Flink流处理系统中进行关联。如何处理？
这里有三种方式：
[*]使用DataStream的API，我们可以使用connect操作，然后再使用RichFlatMapFunction或者是CoProcessFunction来手动实现Join，它们可以将数据保持在state中，在CoProcessFunction中可以使用计时器，定时进行数据的关联，然后定期清理过期的state。
[*]使用Table API，我们自己实现一个UDTF来访问维表
[*]使用流JOIN来实现
我们这里，重点就来讲讲Flink中非常重要的流JOIN操作。在Flink中，流Join主要有两种，一种是Window Join，还有一种是Interval Join。我们先来看看Window Join。
2 Window JOIN
Window Join将流中两个key相同的元素联结在一起。这种联结方式看起来非常像inner join，两个元素必须都存在，才会出现在结果中。
在Flink中，分为有三种不同类型的典型窗口：滚动窗口、滑动窗口、会话窗口。我们以窗口的类型分开讲解。
在执行窗口join时，会将所有key能够匹配上、且处在同一个滚动窗口的事件进行join，join之后传递到JoinFunction或者FlatJoinFunction。这种join看起来就像是INNER JOIN，滚动窗口operator不会将一个在某个流中，而在另一个流中不存在的元素发送到下游。
上述图，表示两个流进行滚动窗口join，我们发现，只要是两个流中都有的元素，才发生了join操作。
来用个简单的案例演示下：
使用两个指定Source模拟数据，一个Source是订单明细，一个Source是商品数据。我们通过window join，将数据关联到一起。
1、先将Flink的依赖导入进来
<repositories>
<repository>
<id>aliyunmaven</id>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</repository>
</repositories>
<properties>
<flink-version>1.12.0</flink-version>
<scala-version>2.12</scala-version>
<mysql-version>5.1.47</mysql-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-version}</artifactId>
<version>${flink-version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala-version}</artifactId>
<version>${flink-version}</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.62</version>
</dependency>
</dependencies>
2、先为本次的测试构建三个实体类，一个是Goods（商品类）、另一个OrderItem（订单明细）、还有一个是Join之后的实体类。
public class Goods11 {
private String goodsId;
private String goodsName;
private BigDecimal goodsPrice;
public static List<Goods11> GOODS_LIST;
public static Random r;
static{
r = new Random();
GOODS_LIST = new ArrayList<>();
GOODS_LIST.add(new Goods11("1", "小米12", new BigDecimal(4890)));
GOODS_LIST.add(new Goods11("2", "iphone12", new BigDecimal(12000)));
GOODS_LIST.add(new Goods11("3", "MacBookPro", new BigDecimal(15000)));
GOODS_LIST.add(new Goods11("4", "Thinkpad X1", new BigDecimal(9800)));
GOODS_LIST.add(new Goods11("5", "MeiZu One", new BigDecimal(3200)));
GOODS_LIST.add(new Goods11("6", "Mate 40", new BigDecimal(6500)));
}
public static Goods11 randomGoods() {
int rIndex = r.nextInt(GOODS_LIST.size());
return GOODS_LIST.get(rIndex);
}
public Goods11() {
}
public Goods11(String goodsId, String goodsName, BigDecimal goodsPrice) {
this.goodsId = goodsId;
this.goodsName = goodsName;
this.goodsPrice = goodsPrice;
}
// 生成getter/setter此处省略
@Override
public String toString() {
return JSON.toJSONString(this);
}
}
public class OrderItem11 {
private String itemId;
private String goodsId;
private Integer count;
// 生成getter/setting，此处省略
@Override
public String toString() {
return JSON.toJSONString(this);
}
}
public class OrderItem11 {
private String itemId;
private String goodsId;
private Integer count;
// 生成getter/setting，此处省略
@Override
public String toString() {
return JSON.toJSONString(this);
}
}
public class FactOrderItem {
private String goodsId;
private String goodsName;
private BigDecimal count;
private BigDecimal totalMoney;
@Override
public String toString() {
return JSON.toJSONString(this);
}
// 生成的getter/setter
}
3、构建一个商品Stream源（这个好比就是维表）
public class GoodsSource11 extends RichSourceFunction {
private Boolean isCancel;
@Override
public void open(Configuration parameters) throws Exception {
isCancel = false;
}
@Override
public void run(SourceContext sourceContext) throws Exception {
while(!isCancel) {
Goods11.GOODS_LIST.stream().forEach(goods -> sourceContext.collect(goods));
TimeUnit.SECONDS.sleep(1);
}
}
@Override
public void cancel() {
isCancel = true;
}
}
4、构建订单明细Stream源
public class OrderItemSource11 extends RichSourceFunction {
private Boolean isCancel;
private Random r;
@Override
public void open(Configuration parameters) throws Exception {
isCancel = false;
r = new Random();
}
@Override
public void run(SourceContext sourceContext) throws Exception {
while(!isCancel) {
Goods11 goods11 = Goods11.randomGoods();
OrderItem11 orderItem11 = new OrderItem11();
orderItem11.setGoodsId(goods11.getGoodsId());
orderItem11.setCount(r.nextInt(10) + 1);
orderItem11.setItemId(UUID.randomUUID().toString());
sourceContext.collect(orderItem11);
orderItem11.setGoodsId("111");
sourceContext.collect(orderItem11);
TimeUnit.SECONDS.sleep(1);
}
}
@Override
public void cancel() {
isCancel = true;
}
}
5、构建水印分配器（此处为了简单），直接使用系统时间了
public class Goods11Watermark implements WatermarkStrategy<Goods11> {
@Override
public TimestampAssigner<Goods11> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
return (element, recordTimestamp) -> System.currentTimeMillis();
}
@Override
public WatermarkGenerator<Goods11> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
return new WatermarkGenerator<Goods11>() {
@Override
public void onEvent(Goods11 event, long eventTimestamp, WatermarkOutput output) {
output.emitWatermark(new Watermark(System.currentTimeMillis()));
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
output.emitWatermark(new Watermark(System.currentTimeMillis()));
}
};
}
}
public class OrderItemWatermark implements WatermarkStrategy<OrderItem11> {
@Override
public TimestampAssigner<OrderItem11> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
return (element, recordTimestamp) -> System.currentTimeMillis();
}
@Override
public WatermarkGenerator<OrderItem11> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
return new WatermarkGenerator<OrderItem11>() {
@Override
public void onEvent(OrderItem11 event, long eventTimestamp, WatermarkOutput output) {
output.emitWatermark(new Watermark(System.currentTimeMillis()));
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
output.emitWatermark(new Watermark(System.currentTimeMillis()));
}
};
}
}
6、Window Join代码
public class TumbleWindowJoin {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 构建商品数据流
SingleOutputStreamOperator<Goods11> goodsDS = env.addSource(new GoodsSource11(), TypeInformation.of(Goods11.class))
.assignTimestampsAndWatermarks(new Goods11Watermark() {
});
// 构建订单明细数据流
SingleOutputStreamOperator<OrderItem11> orderItemDS = env.addSource(new OrderItemSource11(), TypeInformation.of(OrderItem11.class))
.assignTimestampsAndWatermarks(new OrderItemWatermark());
// 进行关联查询
DataStream<FactOrderItem> factOrderItemDS = orderItemDS.join(goodsDS)
// 第一个流orderItemDS
.where(OrderItem11::getGoodsId)
// 第二流goodsDS
.equalTo(Goods11::getGoodsId)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.apply((OrderItem11 item, Goods11 goods) -> {
FactOrderItem factOrderItem = new FactOrderItem();
factOrderItem.setGoodsId(goods.getGoodsId());
factOrderItem.setGoodsName(goods.getGoodsName());
factOrderItem.setCount(new BigDecimal(item.getCount()));
factOrderItem.setTotalMoney(goods.getGoodsPrice().multiply(new BigDecimal(item.getCount())));
return factOrderItem;
});
factOrderItemDS.print();
env.execute("滚动窗口JOIN");
}
}
1、Window Join首先需要使用where和equalTo指定使用哪个key来进行关联，此处我们通过应用方法，基于GoodsId来关联两个流中的元素。
2、设置了5秒的滚动窗口，流的元素关联都会在这个5秒的窗口中进行关联。
3、apply方法中实现了，将两个不同类型的元素关联并生成一个新类型的元素。
3 Interval Join
前面学习的Window Join必须要在一个Window中进行JOIN，那如果没有Window如何处理呢？
interval join也是使用相同的key来join两个流（流A、流B），并且流B中的元素中的时间戳，和流A元素的时间戳，有一个时间间隔。也就是：流B的元素的时间戳 ≥ 流A的元素时间戳 + 下界，且流B的元素的时间戳 ≤ 流A的元素时间戳 + 上界。
我们来看Flink官方的一张图。
我们看到，流A的每一个元素，都会和流B的一定时间范围的元素进行JOIN。
其中，上界和下界可以是负数，也可以是整数。Interval join目前只支持INNER JOIN。将连接后的元素传递给ProcessJoinFunction时，时间戳变为两个元素中最大的那个时间戳。
注意：
Interval Join只支持事件时间。
依然是前面的案例：
public class IntervalJoin {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 构建商品数据流
SingleOutputStreamOperator<Goods11> goodsDS = env.addSource(new GoodsSource11(), TypeInformation.of(Goods11.class))
.assignTimestampsAndWatermarks(new Goods11Watermark() {
});
// 构建订单明细数据流
SingleOutputStreamOperator<OrderItem11> orderItemDS = env.addSource(new OrderItemSource11(), TypeInformation.of(OrderItem11.class))
.assignTimestampsAndWatermarks(new OrderItemWatermark());
// 进行关联查询
SingleOutputStreamOperator<FactOrderItem> factOrderItemDS = orderItemDS.keyBy(item -> item.getGoodsId())
.intervalJoin(goodsDS.keyBy(goods -> goods.getGoodsId()))
.between(Time.seconds(-1), Time.seconds(0))
.upperBoundExclusive()
.process(new ProcessJoinFunction<OrderItem11, Goods11, FactOrderItem>() {
@Override
public void processElement(OrderItem11 left, Goods11 right, Context ctx, Collector<FactOrderItem> out) throws Exception {
FactOrderItem factOrderItem = new FactOrderItem();
factOrderItem.setGoodsId(right.getGoodsId());
factOrderItem.setGoodsName(right.getGoodsName());
factOrderItem.setCount(new BigDecimal(left.getCount()));
factOrderItem.setTotalMoney(right.getGoodsPrice().multiply(new BigDecimal(left.getCount())));
out.collect(factOrderItem);
}
});
factOrderItemDS.print();
env.execute("Interval JOIN");
}
}
我们看到了，上述案例中，我们并没有使用Window就实现了两个流的Join。
1、这里我们通过keyBy将两个流join到一起
2、interval join需要设置流A去关联哪个时间范围的流B中的元素。此处，我设置的下界为-1、上界为0，且上界是一个开区间。表达的意思就是流A中某个元素的时间，对应上一秒的流B中的元素。
3、process中将两个key一样的元素，关联在一起，并加载到一个新的FactOrderItem对象中
参考文献：
https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/operators/joining.html
最新经典文章，欢迎关注公众号http://www.aboutyun.com/data/attachment/forum/201903/18/215536lzpn7n3u7m7u90vm.jpg
---------------------
作者：Flink
来源：weixin
原文：面试实时开发必问——Flink中如何做流Join呢？
页:
[1]