docs

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(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算子传入，可以同时传入多个策略类型；

//指定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包来运行；

首先对应用的源码进行编译

mvn -Prelease-all -DskipTests clean install -U

然后直接通过java指令来运行

 java -jar jarName mainClass

更多详细的案例可以看这里