DataWorks:Origen de datos Kafka

Módulo de datos del registro de Kafka

Tipo de datos procesado

key

Depende del elemento de configuración keyType en la tarea de sincronización de datos. Para obtener más información sobre el parámetro keyType, consulte la descripción completa de parámetros en el apéndice.

value

Depende del elemento de configuración valueType en la tarea de sincronización de datos. Para obtener más información sobre el parámetro valueType, consulte la descripción completa de parámetros en el apéndice.

offset

Long

timestamp

Long

headers

String

partition

Long

Tipo de tarea de sincronización

Formato del value escrito en Kafka

Tipo de campo de origen

Método de procesamiento para operaciones de escritura

Sincronización sin conexión

Nodo de sincronización sin conexión en DataStudio

json

String

Cadena codificada en UTF-8

Boolean

Convertido a la cadena codificada en UTF-8 "true" o "false"

Time/Date

Cadena codificada en UTF-8 con formato yyyy-MM-dd HH:mm:ss

Numeric

Cadena numérica codificada en UTF-8

Flujo de bytes

El flujo de bytes se trata como una cadena codificada en UTF-8 y se convierte a cadena.

text

String

Cadena codificada en UTF-8

Boolean

Convertido a la cadena codificada en UTF-8 "true" o "false"

Time/Date

Cadena codificada en UTF-8 con formato yyyy-MM-dd HH:mm:ss

Numeric

Cadena numérica codificada en UTF-8

Flujo de bytes

El flujo de bytes se trata como una cadena codificada en UTF-8 y se convierte a cadena.

Sincronización en tiempo real: ETL en tiempo real a Kafka

Nodo de sincronización en tiempo real en DataStudio

json

String

Cadena codificada en UTF-8

Boolean

Tipo booleano de JSON

Time/Date

• Para valores de tiempo con precisión inferior a un milisegundo: Se convierten a un entero JSON de 13 dígitos que representa la marca de tiempo en milisegundos.

• Para valores de tiempo con precisión de microsegundos o nanosegundos: Se convierten a un número de punto flotante JSON que incluye un entero de 13 dígitos para la marca de tiempo en milisegundos y un decimal de 6 dígitos para la marca de tiempo en nanosegundos.

Numeric

Tipo numérico de JSON

Flujo de bytes

El flujo de bytes se codifica en Base64 y luego se convierte a una cadena codificada en UTF-8.

text

String

Cadena codificada en UTF-8

Boolean

Convertido a la cadena codificada en UTF-8 "true" o "false"

Time/Date

Cadena codificada en UTF-8 con formato yyyy-MM-dd HH:mm:ss

Numeric

Cadena numérica codificada en UTF-8

Flujo de bytes

El flujo de bytes se codifica en Base64 y luego se convierte a una cadena codificada en UTF-8.

Sincronización en tiempo real: Sincronización en tiempo real de una base de datos completa a Kafka

Sincronización en tiempo real solo de datos incrementales

Formato JSON integrado

String

Cadena codificada en UTF-8

Boolean

Tipo booleano de JSON

Time/Date

Marca de tiempo en milisegundos de 13 dígitos

Numeric

Valor numérico de JSON

Flujo de bytes

El flujo de bytes se codifica en Base64 y luego se convierte a una cadena codificada en UTF-8.

Solución de sincronización: Sincronización en tiempo real a Kafka con un clic

Sincronización completa sin conexión + sincronización incremental en tiempo real

Formato JSON integrado

String

Cadena codificada en UTF-8

Boolean

Tipo booleano de JSON

Time/Date

Marca de tiempo en milisegundos de 13 dígitos

Numeric

Valor numérico de JSON

Flujo de bytes

El flujo de bytes se codifica en Base64 y luego se convierte a una cadena codificada en UTF-8.

Elemento de configuración

Descripción

Módulo de inicio de sesión

Debe establecerse en org.apache.kafka.common.security.plain.PlainLoginModul

username

El nombre de usuario. Configure este elemento según sea necesario.

password

La contraseña. Configure este elemento según sea necesario.

Parámetro

Descripción

Obligatorio

datasource

Nombre del origen de datos. El editor de código permite agregar orígenes de datos. El valor de este parámetro debe coincidir con el nombre del origen de datos agregado.

Sí

server

Dirección del servidor broker de Kafka en formato ip:port.

Solo se puede configurar un server, pero debe asegurarse de que DataWorks pueda conectarse a las direcciones IP de todos los brokers del clúster de Kafka.

Sí

topic

Tópico de Kafka. Un tópico es una agrupación de flujos de mensajes que procesa Kafka.

Sí

column

Datos de Kafka que se leerán. Se admiten columnas constantes, columnas de datos y columnas de atributos.

• Columna constante: columna encerrada entre comillas simples, por ejemplo, ["'abc'", "'123'"].

• Columna de datos

  • Si los datos están en formato JSON, puede obtener las propiedades del objeto JSON, por ejemplo, ["event_id"].

  • Si los datos están en formato JSON, puede obtener las subpropiedades anidadas del objeto JSON, por ejemplo, ["tag.desc"].

• Columna de atributos

  • __key__: clave del mensaje.

  • __value__: contenido completo del mensaje.

  • __partition__: partición donde reside el mensaje actual.

  • __headers__: encabezados del mensaje actual.

  • __offset__: desplazamiento del mensaje actual.

  • __timestamp__: marca de tiempo del mensaje actual.

  A continuación se muestra un ejemplo completo.

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

Sí

keyType

Tipo de la clave de Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG y SHORT.

No

valueType

Tipo del valor de Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG y SHORT.

No

beginDateTime

Hora de inicio para el consumo de datos. Este parámetro especifica el límite izquierdo del intervalo de tiempo, inclusive. Es una cadena de tiempo en formato yyyymmddhhmmss. Puede usar este parámetro junto con los scheduling parameters. Para más información, consulte Supported formats of scheduling parameters.

Debe especificar este parámetro o beginOffset.

endDateTime

Hora de fin para el consumo de datos. Este parámetro especifica el límite derecho del intervalo de tiempo, exclusivo. Es una cadena de tiempo en formato yyyymmddhhmmss. Puede usar este parámetro junto con los scheduling parameters. Para más información, consulte Supported formats of scheduling parameters.

Debe especificar este parámetro o endOffset.

beginOffset

Desplazamiento inicial para el consumo de datos. Puede configurarlo en los siguientes formatos:

• Un número, como 15553274, que indica el desplazamiento inicial para el consumo.

• seekToBeginning: indica que los datos se consumen desde el desplazamiento más antiguo.

• seekToLast: lee datos a partir del desplazamiento guardado para el ID de grupo especificado por group.id en el parámetro kafkaConfig. Tenga en cuenta que el cliente confirma automáticamente el desplazamiento del grupo en el servidor de Kafka a intervalos regulares. Por lo tanto, si una tarea falla y se vuelve a ejecutar, pueden producirse duplicaciones o pérdidas de datos. Si el parámetro skipExceedRecord está establecido en true, la tarea podría descartar los últimos registros leídos. El desplazamiento del grupo correspondiente a estos datos descartados ya se confirmó en el servidor, por lo que no podrán leerse en la siguiente ejecución de la tarea.

• seekToEnd: indica que los datos se consumen desde el desplazamiento más reciente. Esto resultará en la lectura de datos vacíos.

Debe especificar este parámetro o beginDateTime.

endOffset

Desplazamiento final para el consumo de datos. Se utiliza para controlar cuándo finaliza la tarea de consumo de datos.

Debe especificar este parámetro o endDateTime.

skipExceedRecord

Kafka utiliza public ConsumerRecords<K, V> poll(final Duration timeout) para consumir datos. Una sola llamada a poll puede recuperar datos más allá del endOffset o endDateTime especificados. Este parámetro determina si dichos datos excedentes se escriben en el destino. Dado que la tarea utiliza confirmaciones automáticas de desplazamiento, se recomienda lo siguiente:

• Para versiones de Kafka anteriores a la 0.10.2: establezca skipExceedRecord en false.

• Para Kafka 0.10.2 y versiones posteriores: establezca skipExceedRecord en true.

No. El valor predeterminado es false.

partition

Un tópico de Kafka tiene varias particiones (partition). De forma predeterminada, una tarea de sincronización de datos lee datos de un intervalo de desplazamientos que abarca todas las particiones del tópico. También puede especificar una partition para leer datos únicamente del intervalo de desplazamientos de una sola partición.

No. Sin valor predeterminado.

kafkaConfig

Al crear un cliente KafkaConsumer para el consumo de datos, puede especificar parámetros extendidos como bootstrap.servers, auto.commit.interval.ms y session.timeout.ms. Puede usar kafkaConfig para controlar el comportamiento de consumo de KafkaConsumer.

No

encoding

Cuando keyType o valueType se establece en STRING, se utiliza la codificación especificada por este parámetro para analizar sintácticamente la cadena.

No. El valor predeterminado es UTF-8.

waitTIme

Tiempo máximo, en segundos, que el objeto consumidor espera para extraer datos de Kafka en un solo intento.

No. El valor predeterminado es 60.

stopWhenPollEmpty

Los valores válidos son true y false. Si este parámetro se establece en true y el consumidor extrae datos vacíos de Kafka (generalmente porque se han leído todos los datos del tópico, o debido a problemas de red o de disponibilidad del clúster de Kafka), la tarea se detiene inmediatamente. De lo contrario, reintenta hasta que vuelva a leer datos.

No. El valor predeterminado es true.

stopWhenReachEndOffset

Este parámetro solo surte efecto cuando stopWhenPollEmpty es true. Los valores válidos son true y false.

• Si este parámetro se establece en true y el consumidor no recibe datos de una solicitud de sondeo, comprueba si se ha alcanzado el desplazamiento más reciente en las particiones del tópico. Si se ha alcanzado el desplazamiento más reciente en todas las particiones, la tarea se detiene inmediatamente. De lo contrario, continúa intentando extraer datos del tópico.

• Si este parámetro se establece en false y el consumidor no recibe datos de una solicitud de sondeo, no realiza la comprobación y detiene la tarea inmediatamente.

No. El valor predeterminado es false.

Parámetro

Descripción

fetch.min.bytes

Cantidad mínima de datos, en bytes, que el consumidor recupera del broker en una sola solicitud. El broker espera hasta que esta cantidad de datos esté disponible antes de responder al consumidor.

fetch.max.wait.ms

Tiempo máximo, en milisegundos, que el broker espera a que haya datos disponibles antes de responder a una solicitud de recuperación. El valor predeterminado es 500. El broker responde cuando se cumple la condición fetch.min.bytes o fetch.max.wait.ms.

max.partition.fetch.bytes

Especifica el número máximo de bytes que el broker puede devolver al consumidor desde cada partition. El valor predeterminado es 1 MB.

session.timeout.ms

Especifica el tiempo que el consumidor puede estar desconectado del servidor antes de dejar de recibir servicios. El valor predeterminado es 30 segundos.

auto.offset.reset

Acción que realiza el consumidor al leer sin desplazamiento o con un desplazamiento no válido (porque el consumidor ha estado inactivo durante mucho tiempo y el registro con ese desplazamiento ha caducado y se ha eliminado). El valor predeterminado es none, lo que significa que el desplazamiento no se restablece automáticamente. Puede cambiarlo a earliest, lo que significa que el consumidor lee los registros de la partition desde el desplazamiento más antiguo.

max.poll.records

Número de mensajes que puede devolver una sola llamada al método poll.

key.deserializer

Método de deserialización para la clave del mensaje, por ejemplo, org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

Método de deserialización para el valor de los datos, por ejemplo, org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

Ruta del certificado raíz SSL.

ssl.truststore.password

Contraseña del almacén de certificados raíz. Si utiliza Alibaba Cloud Kafka, establézcala en KafkaOnsClient.

security.protocol

Protocolo de acceso. Actualmente, solo se admite el protocolo SASL_SSL.

sasl.mechanism

Método de autenticación SASL. Si utiliza Alibaba Cloud Kafka, use PLAIN.

java.security.auth.login.config

Ruta del archivo de autenticación SASL.

Parámetro

Descripción

Obligatorio

datasource

Nombre del origen de datos. El editor de código permite agregar orígenes de datos. El valor de este parámetro debe coincidir con el nombre del origen de datos agregado.

Sí

server

Dirección del servidor de Kafka en formato ip:port.

Sí

topic

Tópico de Kafka. Es una categoría para los distintos flujos de mensajes que procesa Kafka.

Cada mensaje publicado en un clúster de Kafka tiene una categoría, denominada tópico. Un tópico es una colección de un grupo de mensajes.

Sí

valueIndex

Columna del escritor de Kafka que se utiliza como valor. Si no se especifica, todas las columnas se concatenan para formar el valor de forma predeterminada. El separador se especifica mediante fieldDelimiter.

No

writeMode

Cuando valueIndex no está configurado, este parámetro determina el formato para concatenar todas las columnas del registro de origen y formar el valor del registro de Kafka. Los valores válidos son text y JSON. El valor predeterminado es text.

• Si se establece en text, todas las columnas se concatenan utilizando el separador especificado por fieldDelimiter.

• Si se establece en JSON, todas las columnas se concatenan en una cadena JSON basada en los nombres de campo especificados por el parámetro column.

Por ejemplo, si un registro de origen tiene tres columnas con valores a, b y c, y writeMode se establece en text y fieldDelimiter en #, el valor del registro de Kafka escrito será la cadena a#b#c. Si writeMode se establece en JSON y column en [{"name":"col1"},{"name":"col2"},{"name":"col3"}], el valor del registro de Kafka escrito será la cadena {"col1":"a","col2":"b","col3":"c"}.

Si valueIndex está configurado, este parámetro no tiene efecto.

No

column

Campos de la tabla de destino en los que se escribirán los datos, separados por comas. Por ejemplo: "column": ["id", "name", "age"].

Cuando valueIndex no está configurado y writeMode se establece en JSON, este parámetro define los nombres de campo en la estructura JSON para los valores de columna del registro de origen. Por ejemplo, "column": [{"name":id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

• Si el número de columnas del registro de origen es mayor que el número de nombres de campo configurados en column, los datos se truncan durante la escritura. Por ejemplo:

  Si un registro de origen tiene tres columnas con valores a, b y c, y column se configura como [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}], el valor del registro de Kafka escrito será la cadena {"col1":"a","col2":"b"}.

• Si el número de columnas del registro de origen es menor que el número de campos especificados en column, los campos adicionales se rellenan con null o con la cadena especificada por nullValueFormat. Por ejemplo:

  Si un registro de origen tiene dos columnas con valores a y b, y column se configura como [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}], el valor del registro de Kafka escrito será la cadena {"col1":"a","col2":"b","col3":null}. Si valueIndex está configurado, o si writeMode se establece en text, este parámetro no tiene efecto.

• Si no se configura el tipo de campo JSON, el tipo de campo predeterminado es JSON_STRING.

• Para consultar los valores válidos del tipo de campo JSON, consulte Appendix: JSON field types.

Si valueIndex está configurado, o si writeMode se establece en text, este parámetro no tiene efecto.

Obligatorio cuando valueIndex no está configurado y writeMode se establece en JSON.

partition

Especifica el número de la partición del tópico de Kafka en la que se escribirán los datos. Debe ser un entero mayor o igual que 0.

No

keyIndex

Columna del escritor de Kafka que se utiliza como clave.

El valor del parámetro keyIndex debe ser un entero mayor o igual que 0. De lo contrario, la tarea fallará.

No

keyIndexes

Matriz de los números ordinales de las columnas del registro de origen que se utilizan como clave para el registro de Kafka.

El número ordinal de la columna comienza en 0. Por ejemplo, [0,1,2] concatenará los valores de todos los números de columna configurados con comas para formar la clave del registro de Kafka. Si no se especifica, la clave del registro de Kafka es null y los datos se escriben en las particiones del tópico de forma rotatoria. Solo puede especificar este parámetro o keyIndex.

No

fieldDelimiter

Cuando writeMode se establece en text y valueIndex no está configurado, todas las columnas del registro de origen se concatenan utilizando el separador de columnas especificado por este parámetro para formar el valor del registro de Kafka. Puede configurar un solo carácter o varios caracteres como separador. Puede configurar caracteres Unicode en el formato \u0001. Se admiten caracteres de escape como \t y \n. El valor predeterminado es \t.

Si writeMode no se establece en text o si valueIndex está configurado, este parámetro no tiene efecto.

No

keyType

Tipo de la clave de Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG y SHORT.

Sí

valueType

Tipo del valor de Kafka. Valores válidos: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG y SHORT.

Sí

nullKeyFormat

Si el valor de la columna de origen especificada por keyIndex o keyIndexes es null, se reemplaza por la cadena especificada por este parámetro. Si no se configura, no se realiza ningún reemplazo.

No

nullValueFormat

Si el valor de una columna de origen es null, se reemplaza por la cadena especificada por este parámetro al ensamblar el valor del registro de Kafka. Si no especifica este parámetro, no se realiza ningún reemplazo.

No

acks

Configuración de acks al inicializar el productor de Kafka. Determina el método de confirmación para escrituras exitosas. De forma predeterminada, el parámetro acks se establece en all. Los valores válidos para acks son:

• 0: Sin confirmación para escrituras exitosas.

• 1: Confirmación de escritura exitosa en la réplica principal.

• all: Confirmación de escritura exitosa en todas las réplicas.

No