本文为您介绍MetaQ Reader支持的数据类型、字段映射和数据源等参数及配置示例。
背景信息
消息队列Message Queue(简称MQ)是阿里巴巴集团自主研发的专业消息中间件。消息队列基于高可用分布式集群技术,为您提供消息发布订阅、消息轨迹查询、定时(延时)消息、资源统计和监控报警等消息云服务。消息队列为分布式应用系统提供异步解耦的功能,同时具备海量消息堆积、高吞吐等互联网应用所需要的特性,是阿里巴巴集团双11使用的核心产品。
MetaQ Reader使用消息队列的Java SDK消费消息队列中的实时数据,将数据转换为数据集成传输协议传递给Writer。
实现原理
MetaQ Reader通过消息队列服务的Java SDK订阅MetaQ中的实时消息数据,使用的Java SDK版本如下所示。
<dependency>
<groupId>com.taobao.metaq.final</groupId>
<artifactId>metaq-client</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-sdk</artifactId>
<version>1.3.1</version>
</dependency>
类型转换列表
MetaQ Reader针对MetaQ类型的转换列表,如下所示。
数据集成数据类型 | 消息队列数据类型 |
---|---|
STRING | STRING |
参数说明
参数 | 描述 | 是否必选 |
---|---|---|
accessId | 访问消息队列的访问密钥,用于标识用户。 | 是 |
accessKey | 访问消息队列的访问密钥,用来验证用户的密钥。 | 是 |
consumerId | Consumer是消息的消费者,也称为消息订阅者,负责接收并消费消息。
consumerId是一类Consumer的标识,该类Consumer通常接收并消费一类消息,且消费逻辑一致。 |
是 |
topicName | 消息主题,一级消息类型,通过topic对消息进行分类。 | 是 |
subExpression | 消息子主题。 | 是 |
onsChannel | 用于进行消息队列鉴权。 | 是 |
unitName | 接收消息的目标单元。常用单元如下:
|
否 |
instanceName | Consumer的实例名称。 | 否 |
domainName | 消息队列的接入点。 | 是 |
contentType | 消息的类型,支持singlestringcolumn(消息为STRING类型)、text(消息为文本类型)和json(消息为JSON类型)。 | 是 |
beginOffset | 任务开始读取的Offset,支持begin(从最开始),lastRead(上次读取的Offset)。 | 否 |
nullCurrentOffset | 上次Offset为空时,开始读取的地方。支持begin(从最开始),current(当前offset)。 | 是 |
fieldDelimiter | 分隔符模式下消息字符串的列分隔符,例如逗号等。支持控制字符,例如\u0001。 | 是 |
column | 读取的字段列表。 | 是 |
beginDateTime | 数据消费的开始时间位点,为时间范围(左闭右开)的左边界。
beginDateTime是yyyyMMddHHmmss格式的时间字符串,可以和DataWorks的调度时间参数配合使用。 |
否
说明 beginDateTime和endDateTime配合使用。
|
endDateTime | 数据消费的结束时间位点,为时间范围(左闭右开)的右边界。
endDateTime是yyyyMMddHHmmss格式的时间字符串,可以和DataWorks的调度时间参数配合使用。 |
功能说明
配置一个从消息队列读取数据的示例,使用脚本开发的详情请参见通过脚本模式配置离线同步任务。
{
"job": {
"content": [
{
"reader": {
"name": "metaqreader",
"parameter": {
"accessId": "<yourAccessKeyId>",
"accessKey": "<yourAccessKeySecret>",
"consumerId": "Test01",
"topicName": "test",
"subExpression": "*",
"onsChannel": "ALIYUN",
"domainName": "***.aliyun.com",
"contentType": "singlestringcolumn",
"beginOffset": "lastRead",
"nullCurrentOffset": "begin",
"fieldDelimiter": ",",
"column": [
"col0"
],
"fieldDelimiter": ","
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": false
}
}
}
]
}
}