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

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

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 codificada em UTF-8 "true" ou "false"

Time/Date

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

Numeric

String numérica codificada em UTF-8

Fluxo de bytes

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

text

String

String codificada em UTF-8

Boolean

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

Time/Date

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

Numeric

String numérica codificada em UTF-8

Fluxo de bytes

O fluxo de bytes é tratado como uma string codificada em UTF-8 e convertido em 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 do JSON

Time/Date

• Para valores de tempo com precisão inferior a milissegundos: Convertido em um número inteiro JSON de 13 dígitos que representa o timestamp em milissegundos.

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

Numeric

Tipo numérico do JSON

Fluxo de bytes

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

text

String

String codificada em UTF-8

Boolean

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

Time/Date

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

Numeric

String numérica codificada em UTF-8

Fluxo de bytes

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

Sincronização em tempo real: Sincronização em tempo real de um 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 do JSON

Time/Date

Timestamp de 13 dígitos em milissegundos

Numeric

Valor numérico JSON

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido em 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 do JSON

Time/Date

Timestamp de 13 dígitos em milissegundos

Numeric

Valor numérico JSON

Fluxo de bytes

O fluxo de bytes é codificado em Base64 e depois convertido em 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

O nome de usuário. Configure este item conforme necessário.

password

A senha. Configure este item conforme necessário.

Parâmetro

Descrição

Obrigatório

datasource

O nome da fonte de dados. O editor de código suporta a adição de fontes de dados. O valor deste parâmetro deve ser idêntico ao nome da fonte de dados adicionada.

Sim

server

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

É possível configurar apenas um server, mas é necessário garantir que o DataWorks possa se conectar aos endereços IP de todos os brokers no cluster Kafka.

Sim

topic

O topic do Kafka. Um topic é uma agregação de fluxos de mensagens processados pelo Kafka.

Sim

column

Os dados do Kafka a serem lidos. Colunas constantes, colunas de dados e colunas de atributos são suportadas.

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

• Coluna de dados

  • Se seus dados estiverem no formato JSON, é possível obter as propriedades do objeto JSON, como ["event_id"].

  • Se seus dados estiverem no formato JSON, é possível obter as subpropriedades aninhadas do objeto JSON, como ["tag.desc"].

• Coluna de atributo

  • __key__: A chave da mensagem.

  • __value__: O conteúdo completo da mensagem.

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

  • __headers__: Os cabeçalhos da mensagem atual.

  • __offset__: O offset da mensagem atual.

  • __timestamp__: O timestamp da mensagem atual.

  O código a seguir fornece um exemplo completo.

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

Sim

keyType

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

Não

valueType

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

Não

beginDateTime

A hora inicial para consumo de dados. Este parâmetro especifica o limite esquerdo do intervalo de tempo, que é inclusivo. É uma string de tempo no formato yyyymmddhhmmss. Use este parâmetro com scheduling parameters. Para mais informações, consulte Supported formats of scheduling parameters.

Especifique este parâmetro ou beginOffset.

endDateTime

A hora final para consumo de dados. Este parâmetro especifica o limite direito do intervalo de tempo, que é exclusivo. É uma string de tempo no formato yyyymmddhhmmss. Use este parâmetro com scheduling parameters. Para mais informações, consulte Supported formats of scheduling parameters.

Especifique este parâmetro ou endOffset.

beginOffset

O offset inicial para consumo de dados. Configure-o das seguintes formas:

• Um número, como 15553274, que indica o offset inicial para consumo.

• seekToBeginning: indica que os dados são consumidos a partir do offset mais antigo.

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

• seekToEnd: indica que os dados são consumidos a partir do offset mais recente. Isso resultará na leitura de dados vazios.

Especifique este parâmetro ou beginDateTime.

endOffset

O offset final para consumo de dados. Usado para controlar quando a tarefa de consumo de dados é encerrada.

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ções automáticas de offset, recomendamos o seguinte:

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

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

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

partition

Um topic do Kafka possui múltiplas partições (partition). Por padrão, uma tarefa de sincronização de dados lê dados de um intervalo de offsets que cobre todas as partições do topic. 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, especifique 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 é definido como STRING, a codificação especificada por este parâmetro é usada para analisar a string.

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

waitTIme

O tempo máximo, em segundos, que o objeto consumidor aguarda para extrair dados do Kafka em uma única tentativa.

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

stopWhenPollEmpty

Os valores válidos são true e false. Se este parâmetro for definido como true e o consumidor extrair dados vazios do Kafka (geralmente porque todos os dados no topic foram lidos, ou devido a problemas de rede ou disponibilidade do cluster Kafka), a tarefa para imediatamente. Caso contrário, ela tenta novamente até que os dados sejam lidos.

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

stopWhenReachEndOffset

Este parâmetro só entra em vigor quando stopWhenPollEmpty é true. Os valores válidos são true e false.

• Se este parâmetro for definido como true e o consumidor não receber dados de uma solicitação poll, ele verifica se o offset mais recente nas partições do topic foi atingido. Se o offset mais recente tiver sido atingido para todas as partições, a tarefa para imediatamente. Caso contrário, continua tentando extrair dados do topic.

• Se este parâmetro for definido como false e o consumidor não receber dados de uma solicitação poll, ele não realiza a verificação e interrompe a tarefa imediatamente.

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

Parâmetro

Descrição

fetch.min.bytes

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

fetch.max.wait.ms

O tempo máximo, em milissegundos, que o broker aguarda até que os dados estejam disponíveis antes de responder a uma solicitação de busca. O valor padrão é 500. O broker responde quando a condição fetch.min.bytes ou fetch.max.wait.ms for atendida.

max.partition.fetch.bytes

Especifica o número máximo de bytes que o broker pode retornar ao consumidor de cada partition. O valor padrão é 1 MB.

session.timeout.ms

Especifica o tempo que o consumidor pode ficar desconectado do servidor antes de parar de receber serviços. O valor padrão é 30 segundos.

auto.offset.reset

A ação que o consumidor executa ao ler sem offset ou com um offset inválido (porque o consumidor esteve inativo por muito tempo e o registro com o offset expirou e foi excluído). O valor padrão é none, o que significa que o offset não é redefinido automaticamente. Altere para earliest para que o consumidor leia os registros da partition a partir do offset mais antigo.

max.poll.records

O número de mensagens que podem ser retornadas por uma única chamada ao método poll.

key.deserializer

O método de desserialização para a chave da mensagem, como org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

O método de desserialização para o valor dos dados, como org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

O caminho do certificado raiz SSL.

ssl.truststore.password

A senha para o armazenamento de certificados raiz. Se estiver usando o Alibaba Cloud Kafka, defina como KafkaOnsClient.

security.protocol

O protocolo de acesso. Atualmente, apenas o protocolo SASL_SSL é suportado.

sasl.mechanism

O método de autenticação SASL. Se estiver usando o Alibaba Cloud Kafka, use PLAIN.

java.security.auth.login.config

O caminho do arquivo de autenticação SASL.

Parâmetro

Descrição

Obrigatório

datasource

O nome da fonte de dados. O editor de código suporta a adição de fontes de dados. O valor deste parâmetro deve ser idêntico ao nome da fonte de dados adicionada.

Sim

server

O endereço do servidor Kafka no formato IP:port.

Sim

topic

O topic do Kafka. É uma categoria para diferentes fluxos de mensagens processados pelo Kafka.

Cada mensagem publicada em um cluster Kafka possui uma categoria, chamada de topic. Um topic é uma coleção de um grupo de mensagens.

Sim

valueIndex

A coluna no writer do Kafka usada como valor. Se não for especificado, todas as colunas serão concatenadas para formar o valor por padrão. O separador é especificado por fieldDelimiter.

Não

writeMode

Quando valueIndex não está configurado, este parâmetro determina o formato para concatenar todas as colunas do registro de origem para formar o valor do registro Kafka. Os valores válidos são text e JSON. O valor padrão é text.

• Se definido como text, todas as colunas são concatenadas usando o separador especificado por fieldDelimiter.

• Se definido como JSON, todas as colunas são concatenadas em uma string JSON com base nos nomes de campo especificados pelo parâmetro column.

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

Se valueIndex estiver configurado, este parâmetro será inválido.

Não

column

Os campos na tabela de destino onde os dados serão gravados, separados por vírgulas. Por exemplo: "column": ["id", "name", "age"].

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

• Se o número de colunas no registro de origem for maior que o número de nomes de campo configurados em column, os dados serão truncados durante a gravação. Por exemplo:

  Se um registro de origem tiver três colunas com valores a, b e c, e column estiver configurado como [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}], o valor do registro Kafka gravado será a string {"col1":"a","col2":"b"}.

• Se o número de colunas no registro de origem for menor que o número de campos especificados em column, os campos extras serão preenchidos com null ou a string especificada por nullValueFormat. Por exemplo:

  Se um registro de origem tiver duas colunas com valores a e b, e column estiver configurado como [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}], o valor do registro Kafka gravado será a string {"col1":"a","col2":"b","col3":null}. Se valueIndex estiver configurado, ou se writeMode estiver definido como text, este parâmetro será inválido.

• Se o tipo de campo JSON não estiver configurado, o tipo de campo padrão será JSON_STRING.

• Para os valores válidos do tipo de campo JSON, consulte Appendix: JSON field types.

Se valueIndex estiver configurado, ou se writeMode estiver definido como text, este parâmetro será inválido.

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

partition

Especifica o número da partição no topic do Kafka onde os dados serão gravados. Deve ser um número inteiro maior ou igual a 0.

Não

keyIndex

A coluna no writer do Kafka usada como chave.

O valor do parâmetro keyIndex deve ser um número inteiro maior ou igual a 0. Caso contrário, a tarefa falhará.

Não

keyIndexes

Uma matriz dos números ordinais das colunas no registro de origem que são usadas como chave para o registro Kafka.

O número ordinal da coluna começa em 0. Por exemplo, [0,1,2] concatenará os valores de todos os números de coluna configurados com vírgulas para formar a chave do registro Kafka. Se não for especificado, a chave do registro Kafka será null, e os dados serão gravados nas partições do topic de forma round-robin. Especifique apenas este parâmetro ou keyIndex.

Não

fieldDelimiter

Quando writeMode está definido como text e valueIndex não está configurado, todas as colunas do registro de origem são concatenadas usando o separador de coluna especificado por este parâmetro para formar o valor do registro Kafka. Configure um único caractere ou múltiplos caracteres como separador. Caracteres Unicode podem ser configurados no formato \u0001. Caracteres de escape como \t e \n são suportados. O valor padrão é \t.

Se writeMode não estiver definido como text ou se valueIndex estiver configurado, este parâmetro será inválido.

Não

keyType

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

Sim

valueType

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

Sim

nullKeyFormat

Se o valor da coluna de origem especificada por keyIndex ou keyIndexes for null, ele será substituído pela string especificada neste parâmetro. Se não estiver configurado, nenhuma substituição será feita.

Não

nullValueFormat

Se o valor de uma coluna de origem for null, ele será substituído pela string especificada neste parâmetro ao montar o valor do registro Kafka. Se você não especificar este parâmetro, nenhuma substituição será feita.

Não

acks

A configuração acks ao inicializar o produtor Kafka. Determina o método de confirmação para gravações bem-sucedidas. Por padrão, o parâmetro acks é definido como all. Os valores válidos para acks são:

• 0: Nenhuma confirmação para gravações bem-sucedidas.

• 1: Confirmação para gravação bem-sucedida na réplica primária.

• all: Confirmação para gravação bem-sucedida em todas as réplicas.

Não