DataWorks:Fonte de dados Kafka

Módulo de dados do registro Kafka

Tipo de dados processado

key

Depende do item de configuração keyType na tarefa de sincronização de dados. Para mais informações sobre o parâmetro keyType, consulte a descrição completa dos parâmetros no apêndice.

value

Determinado pelo item de configuração valueType na tarefa de sincronização de dados. Consulte a descrição completa dos parâmetros no apêndice para detalhes sobre valueType.

offset

Long

timestamp

Long

headers

String

partition

Long

Tipo de tarefa de sincronização

Formato do value gravado no Kafka

Tipo de campo de origem

Método de processamento para operações de gravação

Sincronização offline

Nó de sincronização offline no DataStudio

JSON

String

String codificada em UTF-8

Boolean

Convertido para a string UTF-8 "true" ou "false"

Time/Date

String codificada em UTF-8 no formato yyyy-MM-dd HH:mm:ss

Numérico

String numérica codificada em UTF-8

Fluxo de bytes

O fluxo de bytes é tratado como uma string codificada em UTF-8 e convertido para string.

text

String

String codificada em UTF-8

Boolean

Convertido para a string UTF-8 "true" ou "false"

Time/Date

String codificada em UTF-8 no formato yyyy-MM-dd HH:mm:ss

Numérico

String numérica codificada em UTF-8

Fluxo de bytes

O fluxo de bytes é tratado como uma string codificada em UTF-8 e convertido para string.

Sincronização em tempo real: ETL em tempo real para Kafka

Nó de sincronização em tempo real no DataStudio

JSON

String

String codificada em UTF-8

Boolean

Tipo Boolean JSON

Time/Date

• Para valores de tempo com precisão inferior a milissegundos: Convertido para um inteiro JSON de 13 dígitos representando o timestamp em milissegundos.

• Para valores de tempo com precisão de microssegundos ou nanossegundos: Convertido para um número de ponto flutuante JSON contendo um inteiro de 13 dígitos para o timestamp em milissegundos e uma parte decimal de 6 dígitos para o timestamp em nanossegundos.

Numérico

Tipo numérico JSON

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido para uma string codificada em UTF-8.

text

String

String codificada em UTF-8

Boolean

Convertido para a string UTF-8 "true" ou "false"

Time/Date

String codificada em UTF-8 no formato yyyy-MM-dd HH:mm:ss

Numérico

String numérica codificada em UTF-8

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido para uma string codificada em UTF-8.

Sincronização em tempo real: Sincronização em tempo real de banco de dados inteiro para Kafka

Sincronização em tempo real apenas de dados incrementais

Formato JSON integrado

String

String codificada em UTF-8

Boolean

Tipo Boolean JSON

Time/Date

Timestamp de milissegundos com 13 dígitos

Numérico

Valor numérico JSON

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido para uma string codificada em UTF-8.

Solução de sincronização: Sincronização em tempo real com um clique para Kafka

Sincronização offline completa + sincronização incremental em tempo real

Formato JSON integrado

String

String codificada em UTF-8

Boolean

Tipo Boolean JSON

Time/Date

Timestamp de milissegundos com 13 dígitos

Numérico

Valor numérico JSON

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido para uma string codificada em UTF-8.

Item de configuração

Descrição

Módulo de logon

Deve ser definido como org.apache.kafka.common.security.plain.PlainLoginModul

username

Nome de usuário. Configure conforme necessário.

password

Senha. Configure conforme necessário.

Parâmetro

Descrição

Obrigatório

datasource

Nome da fonte de dados. O editor de código permite adicionar fontes de dados. O valor deste parâmetro deve corresponder exatamente ao nome da fonte adicionada.

Sim

server

Endereço do servidor broker Kafka no formato IP:port.

É possível configurar apenas um server, mas garanta que o DataWorks consiga se conectar aos IPs de todos os brokers no cluster Kafka.

Sim

topic

Tópico Kafka. Representa uma agregação de fluxos de mensagens processados pelo Kafka.

Sim

column

Dados do Kafka a serem lidos. Suporta colunas constantes, colunas de dados e colunas de atributos.

• Coluna constante: Coluna entre aspas simples, como ["'abc'", "'123'"].

• Coluna de dados

  • Para dados em formato JSON, é possível extrair propriedades do objeto, como ["event_id"].

  • Para dados JSON, também é possível acessar subpropriedades aninhadas, como ["tag.desc"].

• Coluna de atributo

  • __key__: Chave da mensagem.

  • __value__: Conteúdo completo da mensagem.

  • __partition__: Partição onde a mensagem atual reside.

  • __headers__: Cabeçalhos da mensagem atual.

  • __offset__: Offset da mensagem atual.

  • __timestamp__: Timestamp da mensagem atual.

  O código abaixo fornece um exemplo completo.

  "column": [
      "__key__",
      "__value__",
      "__partition__",
      "__offset__",
      "__timestamp__",
      "'123'",
      "event_id",
      "tag.desc"
      ]

Sim

keyType

Tipo da chave Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG e SHORT.

Não

valueType

Tipo do valor Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG e SHORT.

Não

beginDateTime

Horário inicial para consumo de dados. Define o limite esquerdo (inclusivo) do intervalo de tempo. É uma string no formato yyyymmddhhmmss. Compatível com scheduling parameters. Para mais informações, consulte Formatos suportados de parâmetros de agendamento.

Especifique este parâmetro ou beginOffset.

endDateTime

Horário final para consumo de dados. Define o limite direito (exclusivo) do intervalo de tempo. É uma string no formato yyyymmddhhmmss. Compatível com scheduling parameters. Para mais informações, consulte Formatos suportados de parâmetros de agendamento.

Especifique este parâmetro ou endOffset.

beginOffset

Offset inicial para consumo de dados. Pode ser configurado das seguintes formas:

• Um número, como 15553274, indicando o offset inicial de consumo.

• seekToBeginning: Consome dados a partir do offset mais antigo.

• seekToLast: Lê dados a partir do offset salvo para o group ID definido em group.id no parâmetro kafkaConfig. Note que o offset do grupo é enviado automaticamente ao servidor Kafka pelo cliente em intervalos regulares. Portanto, se uma tarefa falhar e for reexecutada, pode ocorrer duplicação ou perda de dados. Se o parâmetro skipExceedRecord estiver definido como true, a tarefa poderá descartar os últimos registros lidos. Como o offset desses dados já foi confirmado no servidor, eles não serão lidos na próxima execução.

• seekToEnd: Consome dados a partir do offset mais recente. Resultará em leitura de dados vazios.

Especifique este parâmetro ou beginDateTime.

endOffset

Offset final para consumo de dados. Controla quando a tarefa de consumo deve encerrar.

Especifique este parâmetro ou endDateTime.

skipExceedRecord

O Kafka usa public ConsumerRecords<K, V> poll(final Duration timeout) para consumir dados. Uma única chamada poll pode buscar dados além do endOffset ou endDateTime especificado. Este parâmetro determina se esses dados excedentes serão gravados no destino. Como a tarefa usa confirmação automática de offset, recomenda-se:

• Para versões do Kafka anteriores à 0.10.2: Defina skipExceedRecord como false.

• Para Kafka 0.10.2 e posteriores: Defina skipExceedRecord como true.

Não. O valor padrão é false.

partition

Um tópico Kafka possui múltiplas partições (partition). Por padrão, a tarefa lê dados de um intervalo de offsets cobrindo todas as partições do tópico. Também é possível especificar uma partition para ler dados apenas do intervalo de offsets de uma única partição.

Não. Sem valor padrão.

kafkaConfig

Ao criar um cliente KafkaConsumer para consumo de dados, é possível definir parâmetros estendidos como bootstrap.servers, auto.commit.interval.ms e session.timeout.ms. Use kafkaConfig para controlar o comportamento de consumo do KafkaConsumer.

Não

encoding

Quando keyType ou valueType é STRING, a codificação definida aqui é usada para interpretar a string.

Não. O valor padrão é UTF-8.

waitTIme

Tempo máximo, em segundos, que o consumidor aguarda para puxar dados do Kafka em uma única tentativa.

Não. O valor padrão é 60.

stopWhenPollEmpty

Valores válidos: true e false. Se definido como true e o consumidor receber dados vazios do Kafka (geralmente porque todos os dados do tópico foram lidos ou devido a problemas de rede/disponibilidade), a tarefa para imediatamente. Caso contrário, tenta novamente até obter dados.

Não. O valor padrão é true.

stopWhenReachEndOffset

Este parâmetro só tem efeito quando stopWhenPollEmpty é true. Valores válidos: true e false.

• Se definido como true e o consumidor não receber dados numa requisição poll, verifica se o offset mais recente das partições foi atingido. Se todas as partições estiverem no offset mais recente, a tarefa para. Caso contrário, continua tentando puxar dados.

• Se definido como false e não houver dados no poll, a verificação não ocorre e a tarefa para imediatamente.

Não. O valor padrão é false.

Parâmetro

Descrição

fetch.min.bytes

Quantidade mínima de dados, em bytes, que o consumidor busca do broker em uma única requisição. O broker aguarda até ter essa quantidade disponível antes de responder.

fetch.max.wait.ms

Tempo máximo, em milissegundos, que o broker aguarda dados ficarem disponíveis antes de responder a uma requisição fetch. Padrão: 500. O broker responde quando fetch.min.bytes ou fetch.max.wait.ms for satisfeito.

max.partition.fetch.bytes

Define o número máximo de bytes que o broker pode retornar ao consumidor por partition. Padrão: 1 MB.

session.timeout.ms

Tempo que o consumidor pode ficar desconectado do servidor antes de parar de receber serviços. Padrão: 30 segundos.

auto.offset.reset

Ação tomada pelo consumidor ao ler sem offset ou com offset inválido (devido a inatividade prolongada e expiração do registro). Padrão: none (sem redefinição automática). Altere para earliest para ler registros da partition desde o offset mais antigo.

max.poll.records

Número de mensagens retornadas por chamada ao método poll.

key.deserializer

Método de desserialização da chave da mensagem, como org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

Método de desserialização do valor dos dados, como org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

Caminho do certificado raiz SSL.

ssl.truststore.password

Senha do armazenamento de certificados raiz. Para Alibaba Cloud Kafka, use KafkaOnsClient.

security.protocol

Protocolo de acesso. Atualmente, apenas SASL_SSL é suportado.

sasl.mechanism

Método de autenticação SASL. Para Alibaba Cloud Kafka, utilize PLAIN.

java.security.auth.login.config

Caminho do arquivo de autenticação SASL.

Parâmetro

Descrição

Obrigatório

datasource

Nome da fonte de dados. O editor de código permite adicionar fontes de dados. O valor deve corresponder exatamente ao nome da fonte adicionada.

Sim

server

Endereço do servidor Kafka no formato IP:port.

Sim

topic

Tópico Kafka. Categoria para diferentes fluxos de mensagens processados pelo Kafka.

Cada mensagem publicada em um cluster Kafka pertence a uma categoria chamada tópico, que agrupa um conjunto de mensagens.

Sim

valueIndex

Coluna no writer Kafka usada como valor. Se não especificado, todas as colunas são concatenadas para formar o valor, usando o separador definido em fieldDelimiter.

Não

writeMode

Quando valueIndex não está configurado, este parâmetro define o formato de concatenação das colunas do registro de origem para formar o valor do registro Kafka. Valores válidos: text e JSON. Padrão: text.

• Em modo text, as colunas são concatenadas usando o separador de fieldDelimiter.

• Em modo JSON, as colunas formam uma string JSON baseada nos nomes definidos no parâmetro column.

Por exemplo, se um registro de origem tem três colunas com valores a, b e c, writeMode é text e fieldDelimiter é #, o valor gravado será a string a#b#c. Se writeMode for JSON e column for [{"name":"col1"},{"name":"col2"},{"name":"col3"}], o valor será {"col1":"a","col2":"b","col3":"c"}.

Se valueIndex estiver configurado, este parâmetro é ignorado.

Não

column

Campos da tabela de destino onde os dados serão gravados, separados por vírgulas. Exemplo: "column": ["id", "name", "age"].

Quando valueIndex não está configurado e writeMode é JSON, este parâmetro define os nomes dos campos na estrutura JSON para os valores das colunas de origem. Exemplo: "column": [{"name":id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

• Se o número de colunas na origem for maior que o configurado em column, os dados são truncados. Exemplo:

  Se a origem tem três colunas (a, b, c) e column é [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}], o valor gravado será {"col1":"a","col2":"b"}.

• Se a origem tiver menos colunas que o especificado em column, os campos extras recebem null ou a string definida em nullValueFormat. Exemplo:

  Se a origem tem duas colunas (a, b) e column é [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}], o resultado será {"col1":"a","col2":"b","col3":null}. Se valueIndex estiver configurado ou writeMode for text, este parâmetro é ignorado.

• Se o tipo de campo JSON não for definido, assume-se JSON_STRING.

• Para valores válidos de tipos de campo JSON, consulte Apêndice: Tipos de campo JSON.

Se valueIndex estiver configurado ou writeMode for text, este parâmetro é ignorado.

Obrigatório quando valueIndex não está configurado e writeMode é JSON.

partition

Define o número da partição no tópico Kafka onde os dados serão gravados. Deve ser um inteiro maior ou igual a 0.

Não

keyIndex

Coluna no writer Kafka usada como chave.

O valor de keyIndex deve ser um inteiro maior ou igual a 0. Caso contrário, a tarefa falhará.

Não

keyIndexes

Array com os números ordinais das colunas de origem usadas como chave do registro Kafka.

A numeração começa em 0. Por exemplo, [0,1,2] concatena os valores dessas colunas com vírgulas para formar a chave. Se não especificado, a chave será null e os dados serão distribuídos em round-robin pelas partições. Use apenas este parâmetro ou keyIndex.

Não

fieldDelimiter

Quando writeMode é text e valueIndex não está configurado, as colunas de origem são concatenadas usando este separador para formar o valor do registro Kafka. Aceita caracteres únicos ou múltiplos, incluindo Unicode no formato \u0001. Caracteres de escape como \t e \n são suportados. Padrão: \t.

Ignorado se writeMode não for text ou se valueIndex estiver configurado.

Não

keyType

Tipo da chave Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG e SHORT.

Sim

valueType

Tipo do valor Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG e SHORT.

Sim

nullKeyFormat

Se o valor da coluna de origem definida por keyIndex ou keyIndexes for null, ele será substituído pela string especificada aqui. Se não configurado, nenhuma substituição ocorre.

Não

nullValueFormat

Se o valor de uma coluna de origem for null, ele será substituído pela string definida aqui ao montar o valor do registro Kafka. Sem substituição se não especificado.

Não

acks

Configuração acks ao inicializar o producer Kafka. Determina o método de confirmação de gravações bem-sucedidas. Por padrão, acks é all. Valores válidos para acks:

• 0: Sem confirmação de gravação.

• 1: Confirmação de gravação na réplica primária.

• all: Confirmação de gravação em todas as réplicas.

Não