You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ka...@apache.org on 2022/06/21 03:55:04 UTC
[rocketmq-streams] 05/16: merge from upstream/snapshot-1.0.3
This is an automated email from the ASF dual-hosted git repository.
karp pushed a commit to branch snapshot-1.0.4
in repository https://gitbox.apache.org/repos/asf/rocketmq-streams.git
commit a539688121c9a3909af0818b9a01f5347548c256
Author: 维章 <un...@gmail.com>
AuthorDate: Mon May 23 16:09:07 2022 +0800
merge from upstream/snapshot-1.0.3
---
NOTICE | 2 +-
README.md | 16 +--
docs/README.md | 142 +++++++++++++++++++++
docs/SUMMARY.md | 8 ++
...225\264\344\275\223\346\236\266\346\236\204.md" | 33 +++++
.../2.\346\236\204\345\273\272DataStream.md" | 73 +++++++++++
.../3.\345\220\257\345\212\250DataStream.md" | 53 ++++++++
...265\201\350\275\254\350\277\207\347\250\213.md" | 63 +++++++++
...256\227\345\255\220\350\247\243\346\236\220.md" | 55 ++++++++
...256\236\347\216\260\345\256\271\351\224\231.md" | 0
"docs/images/Pipeline\347\261\273\345\233\276.png" | Bin 0 -> 44207 bytes
docs/images/img.png | Bin 38684 -> 0 bytes
docs/images/img_1.png | Bin 43711 -> 0 bytes
docs/images/img_2.png | Bin 103151 -> 0 bytes
docs/images/window.png | Bin 0 -> 241692 bytes
...75\223\346\236\266\346\236\204\345\233\276.png" | Bin 0 -> 60493 bytes
...00\273\344\275\223\350\277\207\347\250\213.png" | Bin 0 -> 44252 bytes
.../\346\211\251\345\256\271\345\211\215.png" | Bin 0 -> 56733 bytes
...12\266\346\200\201\347\256\227\345\255\220.png" | Bin 0 -> 35766 bytes
"docs/images/\347\212\266\346\200\201.png" | Bin 0 -> 47527 bytes
"docs/images/\347\274\251\345\256\271.png" | Bin 0 -> 51087 bytes
quick_start.md => docs/quick_start/README.md | 0
quick_start.md | 92 +++++++++----
.../window/operator/impl/SessionOperator.java | 8 ++
.../streams/window/operator/join/JoinWindow.java | 9 ++
stream_sink.md | 10 +-
stream_source.md | 2 +-
27 files changed, 524 insertions(+), 42 deletions(-)
diff --git a/NOTICE b/NOTICE
index 086ee9fa..a347efb1 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache RocketMQ
-Copyright 2016-2021 The Apache Software Foundation
+Copyright 2016-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/README.md b/README.md
index 51d9a6cf..f2f475e7 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,11 @@
-# Summary
-
+# RocketMQ Streams
+[](https://app.travis-ci.com/apache/rocketmq-streams)
+[](https://app.codecov.io/gh/apache/rocketmq-streams)
+[](https://github.com/apache/rocketmq-streams/releases)
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+[](http://isitmaintained.com/project/apache/rocketmq-streams "Average time to resolve an issue")
+[](http://isitmaintained.com/project/apache/rocketmq-streams "Percentage of issues still open")
+[](https://twitter.com/intent/follow?screen_name=ApacheRocketMQ)
## [中文文档](./README-chinese.md)
@@ -115,9 +121,3 @@ source
.start();
```
-=======
-* [Quick Start](quick\_start.md)
-* [创建实时任务数据源](stream\_source.md)
-* [创建实时任务数据输出](stream\_sink.md)
-* [数据处理逻辑](stream\_transform.md)
->>>>>>> 1cd2dd0291dbcab033e6773021ddca13ce819f82
diff --git a/docs/README.md b/docs/README.md
new file mode 100644
index 00000000..9d5e464e
--- /dev/null
+++ b/docs/README.md
@@ -0,0 +1,142 @@
+[](https://github.com/apache/rocketmq-streams/releases)
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+[](http://isitmaintained.com/project/apache/rocketmq-streams "Average time to resolve an issue")
+[](http://isitmaintained.com/project/apache/rocketmq-streams "Percentage of issues still open")
+[](https://twitter.com/intent/follow?screen_name=ApacheRocketMQ)
+
+# Features
+
+* 轻量级部署：可以单独部署，也支持集群部署
+* 多种类型的数据输入以及输出，source 支持 rocketmq ， sink 支持db, rocketmq 等
+
+# DataStream Example
+
+```java
+import org.apache.rocketmq.streams.client.transform.DataStream;
+
+DataStreamSource source=StreamBuilder.dataStream("namespace","pipeline");
+
+ source
+ .fromFile("～/admin/data/text.txt",false)
+ .map(message->message)
+ .toPrint(1)
+ .start();
+```
+
+# Maven Repository
+
+```xml
+
+<dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-streams-clients</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+# Core API
+
+rocketmq-stream 实现了一系列高级的API，可以让用户很方便的编写流计算的程序，实现自己的业务需求；
+
+## StreamBuilder
+
+StreamBuilder 用于构建流任务的源；
+
++ [dataStream(nameSpaceName,pipelineName)]() 返回DataStreamSource实例，用于分段编程实现流计算任务；
+
+## DataStream API
+
+### Source
+
+DataStreamSource 是分段式编程的源头类，用于对接各种数据源， 从各大消息队列中获取数据；
+
++ ```fromFile``` 从文件中读取数据， 该方法包含俩个参数
+ + ```filePath``` 文件路径，必填参数
+ + ```isJsonData``` 是否json数据， 非必填参数， 默认为```true```
+
+
++ ```fromRocketmq``` 从rocketmq中获取数据，包含四个参数
+ + ```topic``` rocketmq消息队列的topic名称，必填参数
+ + ```groupName``` 消费者组的名称，必填参数
+ + ```isJson``` 是否json格式，非必填参数
+ + ```tags``` rocketmq消费的tags值，用于过滤消息，非必填参数
+
++ ```fromMqtt``` 从满足MQTT协议的终端读取数据， 满足边缘计算的场景，其中包含9个参数
+ + ```url``` mqtt broker的地址，必填参数
+ + ```clientId``` 客户端ID， 必填参数，相同的clientId有负载的作用
+ + ```topic``` topic信息， 必填参数
+ + ```username``` 用户名, 非必填，在mqtt端添加鉴权机制时使用
+ + ```password``` 密码，非必填参数，在mqtt端添加鉴权机制时使用
+ + ```cleanSession``` 是否清理session信息， 非必填，默认为true
+ + ```connectionTimeout``` 连接超时信息， 非必填，默认是10s
+ + ```aliveInterval``` 判断连接是否活跃的间隔时间，非必填，默认是60s
+ + ```automaticReconnect``` 连接断开后自动重连机制，非必填，默认是true
+
+
++ ```from``` 自定义的数据源， 通过实现ISource接口实现自己的数据源
+
+### transform
+
+transform 允许在流计算过程中对输入源的数据进行修改，进行下一步的操作；DataStream API中包括```DataStream```,```JoinStream```, ```SplitStream```,```WindowStream```等多个transform类；
+
+DataStream实现了一系列常见的流计算算子
+
++ ```map``` 通过将源的每个记录传递给函数func来返回一个新的DataStream
++ ```flatmap``` 与map类似，一个输入项对应0个或者多个输出项
++ ```filter``` 只选择func返回true的源DStream的记录来返回一个新的DStream
++ ```forEach``` 对每个记录执行一次函数func， 返回一个新的DataStream
++ ```selectFields``` 对每个记录返回对应的字段值，返回一个新的DataStream
++ ```operate``` 对每个记录执行一次自定义的函数，返回一个新的DataStream
++ ```script``` 针对每个记录的字段执行一段脚本，返回新的字段，生成一个新的DataStream
++ ```toPrint``` 将结果在控制台打印，生成新的DataStream实例
++ ```toFile``` 将结果保存为文件，生成一个新的DataStream实例
++ ```toMqtt``` 将结果输出到满足mqtt协议的设备中，生成一个新的DataStream实例
++ ```toDB``` 将结果保存到数据库
++ ```toRocketmq``` 将结果输出到rocketmq
++ ```to``` 将结果经过自定义的ISink接口输出到指定的存储
++ ```window``` 在窗口内进行相关的统计分析，一般会与```groupBy```连用， ```window()```用来定义窗口的大小， ```groupBy()```用来定义统计分析的主key，可以指定多个
+ + ```count``` 在窗口内计数
+ + ```min``` 获取窗口内统计值的最小值
+ + ```max``` 获取窗口内统计值得最大值
+ + ```avg``` 获取窗口内统计值的平均值
+ + ```sum``` 获取窗口内统计值的加和值
+ + ```reduce``` 在窗口内进行自定义的汇总运算
++ ```join``` 根据条件将俩个流进行内关联
++ ```leftJoin``` 根据条件将俩个流的数据进行左关联
++ ```dimJoin``` 根据条件将流与维表进行内关联，维表的数据可以来自于文件，也可以来自于数据库
++ ```dimLeftJoin``` 根据条件将流与维表进行左关联，维表的数据可以来自于文件，也可以来自于数据库
++ ```union``` 将俩个流进行合并
++ ```split``` 将一个数据流按照标签进行拆分，分为不同的数据流供下游进行分析计算
++ ```with``` with算子用来指定计算过程中的相关策略，包括checkpoint的存储策略，state的存储策略等
+
+#### Strategy
+
+策略机制主要用来控制计算引擎运行过程中的底层逻辑，如checkpoint，state的存储方式等，后续还会增加对窗口、双流join等的控制；所有的控制策略通过```with```算子传入，可以同时传入多个策略类型；
+
+```java
+//指定checkpoint的存储策略
+source
+ .fromRocketmq("TSG_META_INFO","")
+ .map(message->message+"--")
+ .toPrint(1)
+ .with(CheckpointStrategy.db("jdbc:mysql://XXXXX:3306/XXXXX","","",0L))
+ .start();
+```
+
+# 运行
+
+Rocketmq-Streams 作为典型的java应用，既可以集成在业务系统里运行，也可以作为一个独立的jar包来运行；
+
+首先对应用的源码进行编译
+
+```shell
+mvn -Prelease-all -DskipTests clean install -U
+```
+
+然后直接通过java指令来运行
+
+```shell
+ java -jar jarName mainClass
+```
+
+更多详细的案例可以看[这里](docs/SUMMARY.md)
\ No newline at end of file
diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md
new file mode 100644
index 00000000..950f99bd
--- /dev/null
+++ b/docs/SUMMARY.md
@@ -0,0 +1,8 @@
+# Summary
+
+* [Introduction](README.md)
+* [Quick Start](quick_start/README.md)
+* [创建实时任务数据源](stream_source/README.md)
+* [创建实时任务数据输出](stream_sink/README.md)
+* [数据处理逻辑](stream_transform/README.md)
+
diff --git "a/docs/design/1.RocketMQ-streams\346\225\264\344\275\223\346\236\266\346\236\204.md" "b/docs/design/1.RocketMQ-streams\346\225\264\344\275\223\346\236\266\346\236\204.md"
new file mode 100644
index 00000000..8e564a8c
--- /dev/null
+++ "b/docs/design/1.RocketMQ-streams\346\225\264\344\275\223\346\236\266\346\236\204.md"
@@ -0,0 +1,33 @@
+### 总体架构
+
+
+
+数据从RocketMQ中被RocketMQ-streams消费，经过处理最终被写回到RocketMQ。
+如果流处理任务中含有算子groupBy，则需要将数据按照Key进行分组，将分组数据写入shuffle topic。后续算子从
+shuffle topic消费。如果还涉及count之类有状态算子，那么计算时需要读写状态，在窗口触发之后将计算结果写出。
+
+
+### 任务并行度模型
+
+
+
+计算实例实质上是依赖了Rocket-streams SDK的client，因此，计算实例消费的MQ依赖RocketMQ rebalance分配，
+计算实例总个数也不能大于消费总MQ个数，否则将有部分计算实例处于等待状态，消费不到数据。
+
+一个计算实例可以消费多个MQ，一个实例内也只有一张计算拓扑图。
+
+### 状态
+
+
+对于有状态算子，他的状态本地依赖RocksDB加速读取，远程依赖Mysql做持久化。允许流计算任务时，可以只依赖本地存储
+RocksDB, 只需要将setLocalStorageOnly设置成true即可。这种情况下可能存在状态丢失。
+
+
+
+### 扩缩容
+
+
+
+当计算实例从3个缩容到2个，借助于RocketMQ的rebalance，MQ会在计算实例之间重新分配。
+Instance1上消费的MQ2和MQ3被分配到Instance2和Instance3上，这两个MQ的状态数据也需要迁移到Instance2
+和Instance3上，这也暗示，状态数据是根据源数据分片保存的；扩容则是刚好相反的过程。
diff --git "a/docs/design/2.\346\236\204\345\273\272DataStream.md" "b/docs/design/2.\346\236\204\345\273\272DataStream.md"
new file mode 100644
index 00000000..af12e27a
--- /dev/null
+++ "b/docs/design/2.\346\236\204\345\273\272DataStream.md"
@@ -0,0 +1,73 @@
+DataStreamSource中有一个PipelineBuilder，在后续构建过程中，这个PipelineBuilder会一直向后流传，
+将构建过程中产生的source、stage添加进来；最后在start的时候，真正利用PipelineBuilder构建出拓扑图。
+
+### source类型
+ - 设置source的namespace、configureName;
+ - 将source保存到PipelineBuilder中;
+ - 将source作为source节点保存到PipelineBuilder中的ChainPipeline中;
+
+### ChainStage类型
+
+所有的其他运算，包括map,filter,script，window都会先构建出ChainStage，然后以ChainStage的身份进入
+PipelineBuilder，参加后续构建。
+
+在DataStream中一个典型的添加新算子，过程如下所示：
+```java
+
+public DataStream script(String script) {
+ //将用户定义的cript转化成ChainStage
+ // ChainStage<?> stage = this.mainPipelineBuilder.createStage(new ScriptOperator(script));
+ //将ChainStage添加到PipelineBuilder中，构建拓扑。
+ this.mainPipelineBuilder.setTopologyStages(currentChainStage, stage);
+ //将PipelineBuilder构建成DataStream，向后传递，后续还可以用该PipelineBuilder构建拓扑
+ return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, stage);
+}
+
+```
+
+### 创建ChainStage
+
+PipelineBuilder创建，创建过程中会设置label，并将这个ChainStage添加到PipelineBuilder持有ChainPipeline中
+
+- 把ChainStage添加到pipeline中
+ 在构建过程中，所有的添加算子都使用一个共同的PipelineBuilder实例，PipelineBuilder结构如图所示，他持有
+ 一个ChainPipeline实例，ChainPipeline实例中含有一个ISource和多个stages，还有一个label与stage的映射关系，
+ 以及用于寻找下个stage的label。
+ 
+在createStage过程中，将chainStage加入到Pipeline中。
+
+在setTopologyStages 过程中将label加入到Pipeline中；
+
+### 设置拓扑
+```java
+public void setTopologyStages(ChainStage currentChainStage, List<ChainStage> nextStages) {
+ if (isBreak) {
+ return;
+ }
+ if (nextStages == null) {
+ return;
+ }
+ List<String> lableNames = new ArrayList<>();
+ for (ChainStage stage : nextStages) {
+ lableNames.add(stage.getLabel());
+ }
+
+ if (currentChainStage == null) {
+ this.pipeline.setChannelNextStageLabel(lableNames);
+ } else {
+ currentChainStage.setNextStageLabels(lableNames);
+ for (ChainStage stage : nextStages) {
+ stage.getPrevStageLabels().add(currentChainStage.getLabel());
+ }
+ }
+ }
+```
+
+如果是首个ChainStage，则设置下一跳的label；如果不是首个，需要将下个stage的label设置进入当前stage。
+同时，下个stage也需要设置前一个stage的label标签。形成双向链表的结构。
+
+
+
+
+
+
diff --git "a/docs/design/3.\345\220\257\345\212\250DataStream.md" "b/docs/design/3.\345\220\257\345\212\250DataStream.md"
new file mode 100644
index 00000000..91f91f2f
--- /dev/null
+++ "b/docs/design/3.\345\220\257\345\212\250DataStream.md"
@@ -0,0 +1,53 @@
+### Start流程
+
+流式计算在运行时可以拉起多个相同实例进行扩容，所以不能直接启动上述已经构建好的拓扑图，需要将上述构建好的拓扑
+图保存起来，需要扩容时，直接拿出算子的副本，实例化启动即可。
+
+### 统一管理点
+
+- 加载统一管理点IConfigurableService;
+
+ 三种方式存储：Memory, db, file
+
+- PipelineBuilder的build方法，将构建构成中保存起来的IConfigurable，source和statge都是IConfigurable，
+ 保存到IConfigurableService中;
+
+- IConfigurableService的refreshConfigurable方法;
+
+ 1.主要做的事可以概括：从统一管理点加载出组件，赋值，init，在调用后置方法doProcessAfterRefreshConfigurable。
+
+ 2.ChainPipeline的后置方法比较特殊，会调用pipeline中各个组件的后置方法，如果这个组件是普通UDFChainStage，
+ 那么将会反序列化，实例成StageBuilder。如果是WindowChainStage，会讲用户数据接收的window实例化出来。
+
+ 3.从IConfigurable中加载实例副本出来；
+
+ 4.将实例副本赋值；
+
+ 5.初始化实例副本，实例都是AbstractConfigurable的继承类，调用他的的init方法。比如在初始化rocketmqSource
+ 的时候，就会在此时调用init方法，先于启动方法调用；
+
+ 6.调用IConfigurable的doProcessAfterRefreshConfigurable方法，目前只有ChainPipeline会调用，
+ （典型的是ChainPipeline），会在此方法中构建label与stage映射的stageMap；设置source；再调用
+ ChainPipeline中各个stage的doProcessAfterRefreshConfigurable方法；
+
+ 7.这里ChainPipeline的stage都是UDFChainStage类似。UDFChainStage的
+ doProcessAfterRefreshConfigurable方法会将之前序列化好的StageBuilder反序列化，成为StageBuilder实例。
+
+ 8.如果这个stage是window类型的WindowChainStage，ChainPipeline调用各个stage的
+ doProcessAfterRefreshConfigurable。这里会将用于数据接收的window实例化赋值；
+
+ 9.OutputChainStage此时会从统一管理点IConfigurableService查询出sink实例，并赋值给自己sink字段；
+
+
+### ChainPipeline的启动
+```java
+pipeline.startChannel();
+```
+
+将ChainPipeline作为整个数据接收的入口，并启动source；
+
+当source有数据进来时，ChainPipeline将会收到数据；具体方法是ChainPipeline的doMessageInner方法；
+
+该方法将数据封装承AbstractContext后，向后传递；
+
+
diff --git "a/docs/design/4.\346\225\260\346\215\256\347\232\204\346\265\201\350\275\254\350\277\207\347\250\213.md" "b/docs/design/4.\346\225\260\346\215\256\347\232\204\346\265\201\350\275\254\350\277\207\347\250\213.md"
new file mode 100644
index 00000000..1eb791cb
--- /dev/null
+++ "b/docs/design/4.\346\225\260\346\215\256\347\232\204\346\265\201\350\275\254\350\277\207\347\250\213.md"
@@ -0,0 +1,63 @@
+###总体过程
+
+
+
+数据流转整体过程如图所示，黑色箭头线是数据流，橙色为控制流。数据的整体流向是从source中接收到，经过
+AbstractSource判断是否发出系统消息，在进入ChainPipeline,ChainPipeline根据之间构建好的处理拓扑图，使用
+深度优先策略找出下一个处理节点stage，交给Pipeline。Pipeline发现如果是系统消息则对stage执行特殊的控制逻辑，
+如果不是，则用stage来处理具体数据。
+
+### 无window算子执行流程
+- source从RocketMQ中消费数据，进入RocketMQSource的父类AbstractSource；
+- AbstractSource启动控制流，判断是否数据来自新分片，如果是，首先向下游传递一条NewSplitMessage消息，等待系
+ 统消息处理完成返回后，才能继续处理该数据。
+- NewSplitMessage进入Pipeline，如果是系统消息，stage执行该类系统消息对应的控制操作。如果不是系统消息则用
+stage处理数据；
+- Pipeline执行完成后，返回到ChainPipeline，选择下一个stage继续执行；
+- 遍历stage直到结束。
+
+### 含有window算子执行流程
+
+
+
+- 数据流和控制流在上述流程一致，即先进入source，然后由AbstractSource判断是否发出发出系统消息，再进入
+ ChainPipeline按照已经构建好的拓扑图执行。
+- 不同的是，如果是window算子，那么这条数据在执行具体计算之前需要先按照groupBy分组，在执行算子，例如count。
+分组操作需要借助于shuffle topic完成，即写入shuffle topic之前先按照groupBy的值，计算数据写入目的
+ MessageQueue，相同groupBy值的数据将被写入一个MessageQueue中。这样shuffle数据被读取时，
+ groupBy值相同的数据总会被一个client处理，达到按照groupBy分组处理的效果。
+
+- ShuffleChannel会自动订阅、消费shuffle topic。数据会经过shuffle并在ShuffleChannel中再次被消费到。
+- 判断是否是系统消息，如果是，执行该种类系统消息对应的控制流操作。
+- 如果不是系统消息，触发window中算子计算，比如算子是count，就对某个key出现的次数加1；count算子用到的状
+ 态会在接收到NewSplitMessage类型系统消息时提前加载好。计算结束后的状态保存到RocksDB或者mysql中。
+
+- window到时间后，将计算结果输出到下游stage继续计算，并清理RocksDB、Mysql中对应的状态。
+
+
+### 系统消息
+
+#### NewSplitMessage
+当发现数据来自新分片（MessageQueue）时，由AbstractSource产生并向下游拓扑传递。
+
+作用于window算子，使其提前加载该分片对应的状态数据到内存，使得状态数据对该分片数据进行计算时，能使用
+到对应的状态，得出正确的结果。
+
+#### CheckPointMessage
+
+##### 产生时机：
+- 消费分片移除时；
+- RocketMQ-streams向broker提交消费offset时；
+- 处理完一批次消息后；
+
+##### 作用
+- 作用于各个缓存，例如将数据写入shuffle topic之前的WindowCache，使缓存中数据写出到下游。
+- 作用于sink，将sink中缓存而未写出的数据写出；
+- 将有状态算子的状态flush到存储；
+
+#### RemoveSplitMessage
+比较RocketMQ client触发rebalance前后消费的分片，如果某个分片不在被消费，需要将该分片移除，在移除该分配时发出
+RemoveSplitMessage类型消息。
+
+作用于window算子，将RocksDB中状态清除；
+
diff --git "a/docs/design/5.Window\347\256\227\345\255\220\350\247\243\346\236\220.md" "b/docs/design/5.Window\347\256\227\345\255\220\350\247\243\346\236\220.md"
new file mode 100644
index 00000000..4e8be748
--- /dev/null
+++ "b/docs/design/5.Window\347\256\227\345\255\220\350\247\243\346\236\220.md"
@@ -0,0 +1,55 @@
+### window算子初始化
+window的实例化和初始化时机，与普通无状态算子一样，在构建DataStream阶段以stage形式加入pipeline。在启动
+DataStream阶段完成window的初始化。
+
+
+
+- 给window初始化WindowStorage用户状态存储；
+
+ WindowStorage包括localStorage存储和remoteStorage存储；localStorage使用RocksDB，
+ remoteStorage使用mysql；
+
+- 向window添加一个WindowCache的匿名实例，用于存储写入shuffle topic之前数据；
+- 向window添加SQLCache，作为写入Mysql之前的缓存；
+- 向window添加ShuffleChannel，作为写出shuffle和接收来自shufffle topic数据的通道；
+
+
+### ShuffleChannel写出shuffle数据
+AbstractShuffleWindow的doMessage方法，将数据写入shuffleChannel
+```java
+public AbstractContext<IMessage> doMessage(IMessage message, AbstractContext context) {
+ shuffleChannel.startChannel();
+ return super.doMessage(message, context);
+}
+```
+
+- shuffleChannel.startChannel
+启动shuffleChannel中的consumer，从shuffletopic中消费数据；如果有消费到数据，将由
+ shuffleChannel的doMessage处理。
+
+- AbstractWindow.doMessage方法
+
+对于一条消息来说，window 首先需要检查是否有窗口实例，如果没有则创建。如果窗口实例已经超过最大的watermark，
+数据丢弃，否则进行消息积累 消息会先经历batchAdd 然后flush加入到windowCache中；windowCache定时触发，加入到
+shuffleMsgCache中，shuffleMsgCache中定时发出，用shuffleMsgCache中的producer写出到rocketmq。
+
+### ShuffleChannel接收到shuffle数据
+ShuffleChannel#doMessage方法；
+
+将shuffle消息加入到shuffleCache中
+
+最终进入ShuffleCache#batchInsert中
+
+WindowOperator#shuffleCalculate中
+
+实际窗口计算：WindowValue#calculate
+
+计算后并不会马上触发窗口，窗口需要定时出发
+
+### window触发
+ WindowFireSource#startSource启动定时任务，1s检查一次窗口是否触发WindowFireSource#fireWindowInstance
+WindowOperator#fireWindowInstance
+
+windowFireSource.executeMessage
+
+windowFireSource.executeMessage这个方法里面会执行pipeline的下个节点
\ No newline at end of file
diff --git "a/docs/design/6.RocketMQ-streams\345\246\202\344\275\225\345\256\236\347\216\260\345\256\271\351\224\231.md" "b/docs/design/6.RocketMQ-streams\345\246\202\344\275\225\345\256\236\347\216\260\345\256\271\351\224\231.md"
new file mode 100644
index 00000000..e69de29b
diff --git "a/docs/images/Pipeline\347\261\273\345\233\276.png" "b/docs/images/Pipeline\347\261\273\345\233\276.png"
new file mode 100644
index 00000000..dafe81a1
Binary files /dev/null and "b/docs/images/Pipeline\347\261\273\345\233\276.png" differ
diff --git a/docs/images/img.png b/docs/images/img.png
deleted file mode 100644
index b814adf7..00000000
Binary files a/docs/images/img.png and /dev/null differ
diff --git a/docs/images/img_1.png b/docs/images/img_1.png
deleted file mode 100644
index 16a45cc1..00000000
Binary files a/docs/images/img_1.png and /dev/null differ
diff --git a/docs/images/img_2.png b/docs/images/img_2.png
deleted file mode 100644
index 0b75ab05..00000000
Binary files a/docs/images/img_2.png and /dev/null differ
diff --git a/docs/images/window.png b/docs/images/window.png
new file mode 100644
index 00000000..30ba8945
Binary files /dev/null and b/docs/images/window.png differ
diff --git "a/docs/images/\346\200\273\344\275\223\346\236\266\346\236\204\345\233\276.png" "b/docs/images/\346\200\273\344\275\223\346\236\266\346\236\204\345\233\276.png"
new file mode 100644
index 00000000..5eba9cec
Binary files /dev/null and "b/docs/images/\346\200\273\344\275\223\346\236\266\346\236\204\345\233\276.png" differ
diff --git "a/docs/images/\346\200\273\344\275\223\350\277\207\347\250\213.png" "b/docs/images/\346\200\273\344\275\223\350\277\207\347\250\213.png"
new file mode 100644
index 00000000..7a68947e
Binary files /dev/null and "b/docs/images/\346\200\273\344\275\223\350\277\207\347\250\213.png" differ
diff --git "a/docs/images/\346\211\251\345\256\271\345\211\215.png" "b/docs/images/\346\211\251\345\256\271\345\211\215.png"
new file mode 100644
index 00000000..5232b762
Binary files /dev/null and "b/docs/images/\346\211\251\345\256\271\345\211\215.png" differ
diff --git "a/docs/images/\346\234\211\347\212\266\346\200\201\347\256\227\345\255\220.png" "b/docs/images/\346\234\211\347\212\266\346\200\201\347\256\227\345\255\220.png"
new file mode 100644
index 00000000..a9ec479c
Binary files /dev/null and "b/docs/images/\346\234\211\347\212\266\346\200\201\347\256\227\345\255\220.png" differ
diff --git "a/docs/images/\347\212\266\346\200\201.png" "b/docs/images/\347\212\266\346\200\201.png"
new file mode 100644
index 00000000..e2fd9b2e
Binary files /dev/null and "b/docs/images/\347\212\266\346\200\201.png" differ
diff --git "a/docs/images/\347\274\251\345\256\271.png" "b/docs/images/\347\274\251\345\256\271.png"
new file mode 100644
index 00000000..05dcee41
Binary files /dev/null and "b/docs/images/\347\274\251\345\256\271.png" differ
diff --git a/quick_start.md b/docs/quick_start/README.md
similarity index 100%
copy from quick_start.md
copy to docs/quick_start/README.md
diff --git a/quick_start.md b/quick_start.md
index a60dbb95..adcb529d 100644
--- a/quick_start.md
+++ b/quick_start.md
@@ -1,46 +1,84 @@
-# 快速开发
+## rocketmq-streams 快速搭建
+---
-## 引入相关的jar包
+### 前言
-```xml
+本文档主要介绍如何基于rocketmq-streams快速搭建流处理任务，搭建过程中某些例子会用到rocketmq，可以参考[rocketmq搭建文档](https://rocketmq.apache.org/docs/quick-start/)
-<dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-streams-clients</artifactId>
-</dependency>
+### 1、源码构建
-```
+#### 1.1、构建环境
-## 开发实时应用程序
+- JDK 1.8 and above
+- Maven 3.2 and above
-```java
+#### 1.2、构建Rocketmq-streams
-public class RocketmqExample {
+```shell
+git clone https://github.com/apache/rocketmq-streams.git
+cd rocketmq-streams
+mvn clean -DskipTests install -U
- public static void main(String[] args) {
+```
- DataStreamSource dataStream = StreamBuilder.dataStream("test_namespace", "graph_pipeline");
+### 2、基于rocketmq-streams创建应用
- dataStream
- .fromFile("data.csv", false) //构建实时任务的数据源
- .map(message -> message.split(",")) //构建实时任务处理的逻辑过程
- .toPrint(1) //构建实时任务的输出
- .start(); //启动实时任务
- }
-}
+#### 2.1、pom依赖
+```xml
+
+<dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-streams-clients</artifactId>
+</dependency>
```
-## 运行
+#### 2.2、shade clients依赖包
-打包
+```xml
-```shell
-mvn -Prelease-all -DskipTests clean install -U
+<build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.2.1</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <minimizeJar>false</minimizeJar>
+ <shadedArtifactAttached>true</shadedArtifactAttached>
+ <artifactSet>
+ <includes>
+ <include>org.apache.rocketmq:rocketmq-streams-clients</include>
+ </includes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+</build>
```
-运行
+#### 2.3、编写业务代码
+
+快速编写一个统计页面点击次数的小程序：Please see the [rocketmq-streams-examples](rocketmq-streams-examples/README.md)
+
+#### 2.4、运行
+
+- 前提：在从rocketmq中读取数据做流处理时，需要运行topic在rocketmq中自动创建，因为做groupBy操作时，需要用到rocketmq作为shuffle数据的读写目的地。
+- 命令：
-```shell
- java -jar jarName mainClass
```
+ java -jar XXXX-shade.jar \
+ -Dlog4j.level=ERROR \
+ -Dlog4j.home=/logs \
+ -Xms1024m \
+ -Xmx1024m
+```
+
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
index 0c10a9bf..14e1ffd3 100644
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
@@ -524,4 +524,12 @@ public class SessionOperator extends WindowOperator {
}
return numer;
}
+
+ public int getSessionTimeOut() {
+ return sessionTimeOut;
+ }
+
+ public void setSessionTimeOut(int sessionTimeOut) {
+ this.sessionTimeOut = sessionTimeOut;
+ }
}
\ No newline at end of file
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
index f356a286..103f40e3 100644
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
@@ -60,6 +60,7 @@ public class JoinWindow extends AbstractShuffleWindow {
protected String joinType;//join类型，值为INNER,LEFT
protected String expression;//条件表达式。在存在非等值比较时使用
+ protected String rightDependentTableName;
@Override
protected int doFireWindowInstance(WindowInstance instance) {
@@ -557,4 +558,12 @@ public class JoinWindow extends AbstractShuffleWindow {
public void setExpression(String expression) {
this.expression = expression;
}
+
+ public String getRightDependentTableName() {
+ return rightDependentTableName;
+ }
+
+ public void setRightDependentTableName(String rightDependentTableName) {
+ this.rightDependentTableName = rightDependentTableName;
+ }
}
diff --git a/stream_sink.md b/stream_sink.md
index d7718865..a30aae59 100644
--- a/stream_sink.md
+++ b/stream_sink.md
@@ -54,7 +54,7 @@
String topic=.....; //rocketmq 的topic
String namesrvAddress=......; //rocketmq的nameserver
- DataStream dataStream=dataStream.toRocketmq(topic,namesrvAddress);
+ DataStream dataStream=dataStreamSource.toRocketmq(topic,namesrvAddress);
```
@@ -65,7 +65,7 @@
String topic=.....; //rocketmq 的topic
String groupName=.....; // rocketmq的消费组
String namesrvAddress=......; //rocketmq的nameserver
- DataStream dataStream=dataStream.toRocketmq(topic,groupName,namesrvAddress);
+ DataStream dataStream=dataStreamSource.toRocketmq(topic,groupName,namesrvAddress);
```
@@ -77,7 +77,7 @@
String groupName=.....; // rocketmq的消费组
String namesrvAddress=......; //rocketmq的nameserver
String tags=......; // rocketmq的tag信息
- DataStream dataStream=dataStream.toRocketmq(topic,tags,groupName,namesrvAddress);
+ DataStream dataStream=dataStreamSource.toRocketmq(topic,tags,groupName,namesrvAddress);
```
## kafka
@@ -95,7 +95,7 @@
String url=......;
String clientId=......;
String topic=......;
- DataStream dataStream=dataStream.toMqtt(url,cliientId,topic);
+ DataStream dataStream=dataStreamSource.toMqtt(url,cliientId,topic);
```
@@ -108,7 +108,7 @@
String topic=......;
String username=......;
String password=......;
- DataStream dataStream=dataStream.toMqtt(url,cliientId,topic,username,password);
+ DataStream dataStream=dataStreamSource.toMqtt(url,cliientId,topic,username,password);
```
diff --git a/stream_source.md b/stream_source.md
index 8ae7b7e4..77fb96c4 100644
--- a/stream_source.md
+++ b/stream_source.md
@@ -54,7 +54,7 @@
String groupName = .....; // rocketmq的消费组
String namesrvAddress = ......; //rocketmq的nameserver
Boolean isJsonData = true; //是否json
- String tags = ......; //rocketmq的tag信息
+ String tags = ......; // rocketmq的tag信息
DataStream dataStream = dataStreamSource.fromRocketmq(topic, groupName, tags, isJsonData, namesrvAddress);
```