RocketMQ Streams
Features
- 轻量级部署：可以单独部署，也支持集群部署
- 多种类型的数据输入以及输出，source支持 rocketmq ， sink支持db, rocketmq 等
DataStream Example
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
<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()和tableStream()俩个方法，分别返回DataStreamSource和TableStreamSource俩个源；
- dataStream(nameSpaceName,pipelineName) 返回DataStreamSource实例，用于分段编程实现流计算任务；
- tableStream(nameSpaceName,pipelineName)返回TableStreamSource实例， 用于脚本编程实现流计算任务；
DataStream API
Source
DataStreamSource 是分段式编程的源头类，用于对接各种数据源， 从各大消息队列中获取数据；
-
fromFile从文件中读取数据， 该方法包含俩个参数filePath文件路径，必填参数isJsonData是否json数据， 非必填参数， 默认为true
-
fromRocketmq从rocketmq中获取数据，包含四个参数topicrocketmq消息队列的topic名称，必填参数groupName消费者组的名称，必填参数isJson是否json格式，非必填参数tagsrocketmq消费的tags值，用于过滤消息，非必填参数
-
from自定义的数据源， 通过实现ISource接口实现自己的数据源
transform
transform 允许在流计算过程中对输入源的数据进行修改，进行下一步的操作；DataStream API中包括DataStream,JoinStream, SplitStream,WindowStream等多个transform类；
DataStream
DataStream实现了一系列常见的流计算算子
map通过将源的每个记录传递给函数func来返回一个新的DataStreamflatmap与map类似，一个输入项对应0个或者多个输出项filter只选择func返回true的源DStream的记录来返回一个新的DStreamforEach对每个记录执行一次函数func， 返回一个新的DataStreamselectFields对每个记录返回对应的字段值，返回一个新的DataStreamoperate对每个记录执行一次自定义的函数，返回一个新的DataStreamscript针对每个记录的字段执行一段脚本，返回新的字段，生成一个新的DataStreamtoPrint将结果在控制台打印，生成新的DataStreamAction实例toFile将结果保存为文件，生成一个新的DataStreamAction实例toDB将结果保存到数据库toRocketmq将结果输出到rocketmqto将结果经过自定义的ISink接口输出到指定的存储window在窗口内进行相关的统计分析，一般会与groupBy连用，window()用来定义窗口的大小，groupBy()用来定义统计分析的主key，可以指定多个count在窗口内计数min获取窗口内统计值的最小值max获取窗口内统计值得最大值avg获取窗口内统计值的平均值sum获取窗口内统计值的加和值reduce在窗口内进行自定义的汇总运算
join根据条件将将俩个流进行关联， 合并为一个大流进行相关的运算union将俩个流进行合并split将一个数据流按照标签进行拆分，分为不同的数据流供下游进行分析计算withwith算子用来指定计算过程中的相关策略，包括checkpoint的存储策略，state的存储策略等
Strategy
策略机制主要用来控制计算引擎运行过程中的底层逻辑，如checkpoint，state的存储方式等，后续还会增加对窗口、双流join等的控制；所有的控制策略通过with算子传入，可以同时传入多个策略类型；
//指定checkpoint的存储策略
source
.fromRocketmq("TSG_META_INFO","")
.map(message->message+"--")
.toPrint(1)
.with(CheckpointStrategy.db("jdbc:mysql://XXXXX:3306/XXXXX","","",0L))
.start();
45 Apr 4, 2022
23 Nov 20, 2022
9 Nov 10, 2022
2 Dec 14, 2021
49 Dec 31, 2022
3 Feb 1, 2022
3 Dec 18, 2021
6 Feb 22, 2022
2 Jun 5, 2022
4 May 21, 2022
61 Jan 1, 2023
51 Apr 24, 2022
18 Dec 24, 2022
24 Dec 21, 2022