O Data Integration do DataWorks pode usar o MetaQ Reader para ler dados do Message Queue (MQ). Este tópico descreve os recursos de leitura de dados do MetaQ no DataWorks.
Versões suportadas
O MetaQ Reader assina mensagens em tempo real no MetaQ utilizando o SDK Java do Message Queue. As versões do SDK Java são as seguintes:
<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>Limites
Atualmente, você só pode ler dados do MQ usando o plug-in MetaQ Reader no modo de script.
O MetaQ Reader oferece suporte a Serverless resource groups (recommended) e exclusive Data Integration resource groups.
Tipos de campo suportados
O sistema oferece suporte aos seguintes tipos de campo.
|
Tipo de campo |
Leitura offline (MetaQ Reader) |
|
STRING |
Suportado |
O MetaQ Reader mapeia os tipos do MetaQ da seguinte forma.
|
Tipo de dados do Data Integration |
Tipo de dados do Message Queue |
|
STRING |
STRING |
Desenvolver uma tarefa de sincronização de dados
Para obter informações sobre o ponto de entrada e o procedimento de configuração de uma tarefa de sincronização, consulte os seguintes guias de configuração.
Para o procedimento, consulte Script mode configuration.
Para todos os parâmetros e um script de exemplo no modo de script, consulte Appendix: MetaQ sample script and parameter descriptions.
Apêndice: Script de exemplo do MetaQ e descrições de parâmetros
Configurar uma tarefa de sincronização em lote usando o editor de código
Para configure uma tarefa de sincronização em lote usando o editor de código, configure os parâmetros relacionados no script com base nos requisitos unificados de formato de script. Para mais informações, consulte Script mode configuration. As informações a seguir descrevem os parâmetros que você deve configurar para as source de dados ao configure uma tarefa de sincronização em lote pelo editor de código.
Script de exemplo do Reader
{
"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
}
}
}
]
}
}Parâmetros do script do Reader
Parâmetro | Descrição | Obrigatório |
accessId | As informações de AccessKey do Message Queue, usadas para autenticação. | Sim |
accessKey | Sim | |
consumerId | Um consumidor, também chamado de assinante de mensagens, recebe e processa mensagens. consumerIdUm identificador para uma classe de consumidores. Consumidores com o mesmo identificador geralmente recebem e processam o mesmo tipo de mensagem usando a mesma lógica de processamento. | Sim |
topicName | Um tópico de mensagem, que é a categoria principal da mensagem. As mensagens são classificadas por tópico. | Sim |
subExpression | Um subtópico de mensagem. | Sim |
onsChannel | Usado para autenticação com o Message Queue. | Sim |
unitName | A unidade de destino que recebe mensagens. As unidades comuns incluem:
| Não |
instanceName | O nome da instância do consumidor. | Não |
domainName | O endpoint do Message Queue. | Sim |
contentType | O tipo de mensagem. Os valores suportados são singlestringcolumn (uma mensagem STRING), text (uma mensagem de texto) e JSON (uma mensagem JSON). | Sim |
beginOffset | O offset no qual a tarefa inicia a leitura. Os valores suportados são begin (do início) e lastRead (do último offset lido). | Não |
nullCurrentOffset | Onde iniciar a leitura quando o offset anterior estiver vazio. Os valores suportados são begin (do início) e current (do offset atual). | Sim |
fieldDelimiter | O delimitador de coluna para strings de mensagens no modo delimitador, como uma vírgula. Há suporte para caracteres de controle, por exemplo \u0001. | Sim |
column | A lista de campos a serem lidos. | Sim |
beginDateTime | A hora de início para o consumo de dados, que é o limite inferior inclusivo do intervalo de tempo. beginDateTime é uma string de carimbo de data/hora no formato yyyyMMddHHmmss e pode ser usada com os parâmetros de tempo de agendamento do DataWorks. | Não Nota beginDateTime e endDateTime são usados em conjunto. |
endDateTime | A hora de término para o consumo de dados, que é o limite superior exclusivo do intervalo de tempo. endDateTime é uma string de carimbo de data/hora no formato yyyyMMddHHmmss e pode ser usada com os parâmetros de tempo de agendamento do DataWorks. |