使用方式,

dataStream.coGroup(otherStream)
.where(0).equalTo(1)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.apply (new CoGroupFunction () {...});

 

可以看到coGroup只是产生CoGroupedStreams

    public <T2> CoGroupedStreams<T, T2> coGroup(DataStream<T2> otherStream) {
return new CoGroupedStreams<>(this, otherStream);
}

 

而where, equalTo只是添加keySelector,对于两个流需要分别指定

keySelector1,keySelector2

 

window设置双流的窗口,很容易理解

 

apply,

       /**
* Completes the co-group operation with the user function that is executed
* for windowed groups.
*
* <p>Note: This method's return type does not support setting an operator-specific parallelism.
* Due to binary backwards compatibility, this cannot be altered. Use the
* {@link #with(CoGroupFunction, TypeInformation)} method to set an operator-specific parallelism.
*/
public <T> DataStream<T> apply(CoGroupFunction<T1, T2, T> function, TypeInformation<T> resultType) {
//clean the closure
function = input1.getExecutionEnvironment().clean(function); UnionTypeInfo<T1, T2> unionType = new UnionTypeInfo<>(input1.getType(), input2.getType());
UnionKeySelector<T1, T2, KEY> unionKeySelector = new UnionKeySelector<>(keySelector1, keySelector2); DataStream<TaggedUnion<T1, T2>> taggedInput1 = input1 //将input1封装成TaggedUnion,很简单,就是赋值到one上
.map(new Input1Tagger<T1, T2>())
.setParallelism(input1.getParallelism())
.returns(unionType);
DataStream<TaggedUnion<T1, T2>> taggedInput2 = input2 //将input2封装成TaggedUnion
.map(new Input2Tagger<T1, T2>())
.setParallelism(input2.getParallelism())
.returns(unionType); DataStream<TaggedUnion<T1, T2>> unionStream = taggedInput1.union(taggedInput2); //由于现在双流都是TaggedUnion类型,union成一个流,问题被简化 // we explicitly create the keyed stream to manually pass the key type information in
WindowedStream<TaggedUnion<T1, T2>, KEY, W> windowOp = //创建窗口
new KeyedStream<TaggedUnion<T1, T2>, KEY>(unionStream, unionKeySelector, keyType)
.window(windowAssigner); if (trigger != null) { //如果有trigger,evictor,设置上
windowOp.trigger(trigger);
}
if (evictor != null) {
windowOp.evictor(evictor);
} return windowOp.apply(new CoGroupWindowFunction<T1, T2, T, KEY, W>(function), resultType); //调用window的apply
}

关键理解,他要把两个流变成一个流,这样问题域就变得很简单了

最终调用到WindowedStream的apply,apply是需要保留window里面的所有原始数据的,和reduce不一样

apply的逻辑,是CoGroupWindowFunction

 

private static class CoGroupWindowFunction<T1, T2, T, KEY, W extends Window>
extends WrappingFunction<CoGroupFunction<T1, T2, T>>
implements WindowFunction<TaggedUnion<T1, T2>, T, KEY, W> { private static final long serialVersionUID = 1L; public CoGroupWindowFunction(CoGroupFunction<T1, T2, T> userFunction) {
super(userFunction);
} @Override
public void apply(KEY key,
W window,
Iterable<TaggedUnion<T1, T2>> values,
Collector<T> out) throws Exception { List<T1> oneValues = new ArrayList<>();
List<T2> twoValues = new ArrayList<>(); for (TaggedUnion<T1, T2> val: values) {
if (val.isOne()) {
oneValues.add(val.getOne());
} else {
twoValues.add(val.getTwo());
}
}
wrappedFunction.coGroup(oneValues, twoValues, out);
}
}
}

逻辑也非常的简单,就是将该key所在window里面的value,放到oneValues, twoValues两个列表中

最终调用到用户定义的wrappedFunction.coGroup

 

DataStream.join就是用CoGroup实现的

            return input1.coGroup(input2)
.where(keySelector1)
.equalTo(keySelector2)
.window(windowAssigner)
.trigger(trigger)
.evictor(evictor)
.apply(new FlatJoinCoGroupFunction<>(function), resultType);

 

FlatJoinCoGroupFunction

private static class FlatJoinCoGroupFunction<T1, T2, T>
extends WrappingFunction<FlatJoinFunction<T1, T2, T>>
implements CoGroupFunction<T1, T2, T> {
private static final long serialVersionUID = 1L; public FlatJoinCoGroupFunction(FlatJoinFunction<T1, T2, T> wrappedFunction) {
super(wrappedFunction);
} @Override
public void coGroup(Iterable<T1> first, Iterable<T2> second, Collector<T> out) throws Exception {
for (T1 val1: first) {
for (T2 val2: second) {
wrappedFunction.join(val1, val2, out);
}
}
}
}

可以看出当前join是inner join,必须first和second都有的情况下,才会调到用户的join函数

Flink - CoGroup的更多相关文章

  1. Alink漫谈(十六) :Word2Vec源码分析 之 建立霍夫曼树

    Alink漫谈(十六) :Word2Vec源码分析 之 建立霍夫曼树 目录 Alink漫谈(十六) :Word2Vec源码分析 之 建立霍夫曼树 0x00 摘要 0x01 背景概念 1.1 词向量基础 ...

  2. Flink学习笔记:Operators之CoGroup及Join操作

    本文为<Flink大数据项目实战>学习笔记,想通过视频系统学习Flink这个最火爆的大数据计算框架的同学,推荐学习课程: Flink大数据项目实战:http://t.cn/EJtKhaz ...

  3. Flink入门 - CoGroup和Join

    /* *CoGroup */ final StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironm ...

  4. Flink实例(五十): Operators(十)多流转换算子(五)coGroup 与union

    参考链接:https://mp.weixin.qq.com/s/BOCFavYgvNPSXSRpBMQzBw 需求场景分析 需求场景 需求诱诱诱来了...数据产品妹妹想要统计单个短视频粒度的「点赞,播 ...

  5. Flink Program Guide (2) -- 综述 (DataStream API编程指导 -- For Java)

    v\:* {behavior:url(#default#VML);} o\:* {behavior:url(#default#VML);} w\:* {behavior:url(#default#VM ...

  6. Flink Program Guide (1) -- 基本API概念(Basic API Concepts -- For Java)

    false false false false EN-US ZH-CN X-NONE /* Style Definitions */ table.MsoNormalTable {mso-style-n ...

  7. Flink从入门到放弃(入门篇3)-DataSetAPI

    戳更多文章: 1-Flink入门 2-本地环境搭建&构建第一个Flink应用 3-DataSet API 4-DataSteam API 5-集群部署 6-分布式缓存 7-重启策略 8-Fli ...

  8. Flink Java Demo(Windows)

    关于Flink相关的概念性东西就不说了,网上都有,官网也很详尽.本文主要记录一下Java使用Flink的简单例子. 首先,去官网下载Flink的zip包(链接就不提供了,你已经是个成熟的程序员了,该有 ...

  9. Flink官网文档翻译

    http://ifeve.com/flink-quick-start/ http://vinoyang.com/2016/05/02/flink-concepts/ http://wuchong.me ...

随机推荐

  1. ImageSharp .NET Core跨平台图形处理库

    ImageSharp 是支持.NET Core跨平台图形处理库,ImageSharp是ImageProcessor 的.NET Core跨平台实现. ImageSharp 支持如下操作: 调整大小,裁 ...

  2. Codeforces Beta Round #83 (Div. 1 Only)题解【ABCD】

    Codeforces Beta Round #83 (Div. 1 Only) A. Dorm Water Supply 题意 给你一个n点m边的图,保证每个点的入度和出度最多为1 如果这个点入度为0 ...

  3. lunix机器的jdk安装

    本来不想写这篇博客的,写在这儿只是作为自己的笔记,jdk安装了千万编,但是踩过的坑老是不记,看别人的博客又各种不爽,所有索性自己写一个得了.老规矩,无图. Oracle版本的jdk下载地址:http: ...

  4. laravel框架——表单验证

    创建路由 Route::get('test','VerController@index'); Route::post('tosubmit','VerController@tosubmit'); 在控制 ...

  5. d3可视化实战03:神奇的superformula

    需求驱动实现 前文讲过了D3的数据驱动机制,中间所举的例子都很简单.例如那个demo里面,绑定的数据是一个简单的数组,实现的图元也仅仅是一堆用SVG画的circle.但是现实世界中我们往往会遇到复杂的 ...

  6. python wmi模块学习

    # -*- coding: cp936 -*- import wmi c = wmi.WMI () for sys in c.Win32_OperatingSystem(): print " ...

  7. 删除字符串中多余的空白字符和空行(C语言实现)

    要求:处理一个字符串,删除字符串中多余的空格.水平制表符和空行,并满足下列要求: (1)对原字符串只能进行一次扫描.(2)不允许申请新的空间.(3)处理后的字符串的首尾不能有空格.制表符和空行.(4) ...

  8. Python递归函数

    参考: https://pythonspot.com/recursion/ https://www.python-course.eu/recursive_functions.php 一.递归函数两大要 ...

  9. Python内置函数(8)——bytes

    英文文档: class bytes([source[, encoding[, errors]]]) Return a new “bytes” object, which is an immutable ...

  10. webgl,threejs教程、笔记

    发现一个不错的博客,学学. webgl和threejs教程