Realtime Compute for Apache Flink:Kafka connector

Descripción general

Requisitos previos

Notas

No se recomiendan las escrituras transaccionales debido a limitaciones de diseño conocidas en Apache Flink y Apache Kafka. Cuando establece sink.delivery-guarantee = 'exactly-once', el conector de Kafka habilita las escrituras transaccionales, lo cual presenta los siguientes problemas conocidos:

• Cada checkpoint genera un nuevo Transaction ID. Si el intervalo entre checkpoints es demasiado corto, la gran cantidad de Transaction IDs resultante puede agotar la memoria del coordinador del clúster de Kafka y comprometer su estabilidad.

• Cada transacción crea una nueva instancia de Producer. Si se confirman demasiadas transacciones de forma simultánea, el TaskManager puede quedarse sin memoria y desestabilizar el trabajo de Apache Flink.

• Si varios trabajos de Apache Flink usan el mismo sink.transactional-id-prefix, los Transaction IDs generados pueden entrar en conflicto. Cuando falla una operación de escritura en un trabajo, se puede impedir que avance el Log Start Offset (LSO) de una partición de Apache Kafka. Esto afecta a todos los consumidores de esa partición.

Si necesita semántica exactly-once, utilice Upsert Kafka connector para escribir en una tabla con clave primaria y así garantizar la idempotencia. Si es indispensable usar escrituras transaccionales, consulte Exactly-once semantics usage notes.

Solución de problemas de conectividad de red

Si un trabajo de Realtime Compute for Apache Flink no inicia y muestra el error Timed out waiting for a node assignment, suele indicar un problema de conectividad de red entre Realtime Compute for Apache Flink y el clúster de Kafka.

SQL

Utilice el conector de Kafka como tabla de origen o tabla de destino en trabajos SQL.

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

Columnas de metadatos

CREATE TABLE kafka_source (
  -- Read the message topic as the `record_topic` column
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` column
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the message offset as the `record_offset` column
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Write the timestamp from the `ts` column as the ProducerRecord's timestamp to Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

Opciones del conector

• Generales

  Option	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  connector	El tipo de conector.	String	Sí	–	El valor debe ser kafka.
  properties.bootstrap.servers	Una lista de direcciones de brokers de Kafka.	String	Sí	–	Formato: host1:port1,host2:port2,.... Separe las direcciones con comas (,).
  properties.*	Propiedades adicionales para el cliente de Kafka.	String	No	–	Las claves de propiedad deben ser opciones válidas definidas en la documentación oficial de Apache Kafka para Producer Configs y Consumer Configs. Realtime Compute for Apache Flink elimina el prefijo properties. y pasa los pares clave-valor restantes al cliente de Kafka subyacente. Por ejemplo, establezca 'properties.allow.auto.create.topics' = 'false' para deshabilitar la creación automática de temas. El conector de Kafka sobrescribe estas opciones, por lo que no se pueden configurar de esta manera: key.deserializer value.deserializer
  format	El formato para serializar y deserializar el valor de un mensaje de Kafka.	String	No	–	Formatos compatibles: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Nota Para obtener más información, consulte Opciones de formato.
  key.format	El formato para serializar y deserializar la clave de un mensaje de Kafka.	String	No	–	Formatos compatibles: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Nota Cuando utilice esta configuración, key.options es obligatorio.
  key.fields	Los campos del esquema de la tabla que se usan como clave del mensaje de Kafka.	String	No	–	Separe varios nombres de campo con puntos y comas (;). Por ejemplo, 'field1;field2'.
  key.fields-prefix	Un prefijo personalizado para todos los campos de clave, con el fin de evitar conflictos de nombre con los campos de valor.	String	No	–	Este prefijo distingue los campos de clave de los de valor. Se elimina antes de serializar la clave o después de deserializarla. Nota Si utiliza esta opción, value.fields-include debe establecerse en EXCEPT_KEY.
  value.format	El formato para serializar y deserializar el valor de un mensaje de Kafka.	String	No	–	Esta configuración equivale a format. Solo puede establecer format o value.format, pero no ambos. Si configura los dos, value.format anula a format.
  value.fields-include	Define si los campos de clave se incluyen en el formato de valor.	String	No	ALL	Valores válidos: ALL: El valor del mensaje de Kafka incluye todas las columnas de la tabla. EXCEPT_KEY: El valor del mensaje de Kafka incluye todas las columnas de la tabla, excepto las definidas en key.fields.

• Tabla de origen

  Option	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  topic	El tema o los temas de los que se leerá.	String	No	–	Para suscribirse a varios temas, separe sus nombres con puntos y comas (;); por ejemplo, 'topic-1;topic-2'. Nota Puede especificar esta opción o topic-pattern, pero no ambas.
  topic-pattern	Una expresión regular que coincide con los temas a los que desea suscribirse. El consumidor se suscribe a todos los temas cuyos nombres coincidan con este patrón.	String	No	–	Ejemplos: user_event_.*: Coincide con todos los temas que tienen el prefijo user_event_. prod\.logs\..*: Coincide con los temas que tienen el prefijo prod.logs. (el carácter . debe escaparse). Nota Puede especificar esta opción o topic, pero no ambas.
  properties.group.id	El ID del grupo de consumidores de origen de Kafka.	String	No	KafkaSource-{Source-Table-Name}	Si utiliza un ID de grupo de consumidores por primera vez, también debe establecer properties.auto.offset.reset en earliest o latest para definir el desplazamiento inicial de arranque.
  scan.startup.mode	El desplazamiento de arranque del consumidor de Kafka.	String	No	group-offsets	Valores válidos: earliest-offset: Comienza a leer desde el desplazamiento más antiguo disponible. latest-offset: Comienza a leer desde el desplazamiento más reciente. group-offsets: Comienza a leer desde los desplazamientos confirmados del properties.group.id especificado. timestamp: Comienza a leer desde el scan.startup.timestamp-millis especificado. specific-offsets: Comienza a leer desde los desplazamientos especificados en scan.startup.specific-offsets. Nota Esta opción solo aplica cuando un trabajo inicia sin estado. Si un trabajo se reanuda desde un checkpoint, lee desde los desplazamientos almacenados en el estado del checkpoint.
  scan.startup.specific-offsets	El desplazamiento de inicio por partición cuando scan.startup.mode es specific-offsets.	String	No	–	Por ejemplo, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	La marca de tiempo de inicio en milisegundos cuando scan.startup.mode se establece en timestamp.	Long	No	–	La unidad es milisegundos.
  scan.topic-partition-discovery.interval	El intervalo de detección de particiones.	Duration	No	5 minutos	El conector detecta periódicamente nuevas particiones y lee de ellas. Cuando utiliza topic-pattern, el conector también detecta nuevos temas que coinciden con el patrón. Establezca el intervalo en un valor no positivo para deshabilitar esta función. Nota En Ververica Runtime (VVR) 6.0.x, la detección dinámica de particiones está deshabilitada de forma predeterminada. A partir de VVR 8.0, esta función se habilita de forma predeterminada con un intervalo de detección de 5 minutos.
  scan.header-filter	Filtra mensajes según los encabezados de los mensajes de Kafka.	String	No	–	Una clave de encabezado y su valor se separan con dos puntos (:). Varias condiciones de encabezado se conectan mediante operadores lógicos (& y |). También se admite el operador lógico NOT (!). Por ejemplo, depart:toy|depart:book&!env:test conserva los datos de Kafka si el encabezado contiene depart=toy o depart=book y no contiene env=test. Nota Esta opción solo es compatible con Ververica Runtime (VVR) 8.0.6 y versiones posteriores. No se admiten paréntesis en las expresiones. Las operaciones lógicas se evalúan de izquierda a derecha. Los valores de los encabezados se convierten a cadenas UTF-8 para la comparación.
  scan.check.duplicated.group.id	Comprueba si otro consumidor activo ya está utilizando el properties.group.id.	Boolean	No	false	Valores válidos: true: Antes de iniciar el trabajo, el sistema comprueba si existe un grupo de consumidores duplicado. Si encuentra uno, el trabajo falla para evitar conflictos. false: Inicia el trabajo sin comprobar si hay conflictos. Nota Esta opción solo es compatible con Ververica Runtime (VVR) 6.0.4 y versiones posteriores.

• Tabla de destino

  Option	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  topic	El tema de destino.	String	Sí	–	–
  sink.partitioner	Asigna registros de instancias de destino paralelas a particiones de Kafka.	String	No	default	Valores válidos: default: Utiliza el particionador predeterminado de Kafka. fixed: Cada instancia de destino paralela escribe en una partición fija de Kafka. round-robin: Los registros se distribuyen entre las particiones de forma rotativa. Particionador personalizado: Para usar un particionador personalizado, proporcione el nombre de clase completo de una subclase de FlinkKafkaPartitioner; por ejemplo, org.mycompany.MyPartitioner.
  sink.delivery-guarantee	La garantía de entrega del destino.	String	No	at-least-once	Valores válidos: none: No ofrece garantías. Los registros pueden perderse o duplicarse. at-least-once: Garantiza que no se pierdan registros, aunque podrían duplicarse. exactly-once: Utiliza transacciones de Kafka para ofrecer semántica exactly-once, lo que garantiza que los registros no se pierdan ni se dupliquen. Nota Cuando utilice la semántica exactly-once, también debe especificar sink.transactional-id-prefix.
  sink.transactional-id-prefix	El prefijo del ID de transacción. Es obligatorio cuando sink.delivery-guarantee es exactly-once.	String	Sí, si sink.delivery-guarantee es exactly-once	–	Solo es obligatorio cuando sink.delivery-guarantee se establece en exactly-once.
  sink.parallelism	El paralelismo del operador de destino.	Integer	No	–	De forma predeterminada, el framework determina el paralelismo según los operadores anteriores.

Seguridad y autenticación

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* SSL configuration */
  /* Path to the truststore for the server's CA certificate. */
  /* Files uploaded using Artifacts are stored in the /flink/usrlib/ directory. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* If client authentication is required, you must also configure the path to the keystore (private key). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* The algorithm used to verify the server hostname. An empty string disables hostname verification. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* SASL configuration */
  /* Set the SASL mechanism to SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Configure JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Utilice la función Artifacts de la consola de Realtime Compute for Apache Flink para cargar el certificado de CA y la clave privada mencionados en el ejemplo. Los archivos cargados se almacenan en el directorio /flink/usrlib. Para usar un archivo de certificado de CA llamado my-truststore.jks, establezca la propiedad 'properties.ssl.truststore.location' en la cláusula WITH de una de las siguientes dos maneras:

• Establezca 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'. Este método evita descargar archivos dinámicamente desde Object Storage Service (OSS) durante la ejecución, pero no es compatible con Debug Mode.

• Si la versión del motor de Realtime Compute es VVR 11.5 o posterior, configure properties.ssl.truststore.location y properties.ssl.keystore.location con una ruta absoluta de OSS. El formato de la ruta del archivo es oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Namespace name>/<file name>. Este método descarga dinámicamente los archivos de OSS durante la ejecución de Flink y es compatible con Debug Mode.

Desplazamiento de inicio de la tabla de origen

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume from the committed offsets of the consumer group "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- If "my-group" is used for the first time, consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consumption starts from the latest offset.
  -- Consume from the specified timestamp in milliseconds: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from specific offsets.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Confirmación de desplazamientos de origen

La tabla de origen de Kafka confirma su desplazamiento de consumo en el clúster de Kafka únicamente después de un checkpoint exitoso. Por ello, un intervalo largo entre checkpoints provoca un retraso en el desplazamiento confirmado. La tabla de origen almacena el progreso real de lectura en el estado del checkpoint, el cual utiliza el sistema para la recuperación ante fallos. Los desplazamientos confirmados sirven solo como monitor de progreso y no se utilizan para la recuperación, por lo que los errores de confirmación no afectan la precisión de los datos.

Particionador de sink personalizado

Si la estrategia de particionamiento integrada de Kafka no cumple con sus requisitos, implemente un particionador personalizado mediante la extensión de la clase FlinkKafkaPartitioner. Una vez finalizado el desarrollo, compile su código en un paquete JAR y cárguelo con la función Artifacts de la consola de Realtime Compute. Después de cargar y referenciar el paquete JAR, establezca el parámetro sink.partitioner de la cláusula WITH con el nombre de clase completo de su particionador; por ejemplo, org.mycompany.MyPartitioner.

Kafka, Upsert Kafka y catálogo JSON de Kafka

Kafka es una plataforma de streaming de eventos de solo anexión que no admite actualizaciones ni eliminaciones de datos. En SQL de streaming, una tabla sink estándar de Kafka no procesa datos de Change Data Capture (CDC) provenientes del flujo superior ni la lógica de retracción de operadores como aggregate y join. Si necesita escribir datos que contengan cambios o retracciones, utilice una tabla sink Upsert Kafka.

Para simplificar la sincronización por lotes de datos de Change Data Capture (CDC) desde una o más tablas de base de datos superiores hacia Kafka, utilice un catálogo JSON de Kafka. Cuando los datos almacenados en Kafka tienen formato JSON, este catálogo le permite omitir la definición del esquema y de los parámetros WITH. Para obtener más detalles, consulte Manage Kafka JSON catalogs.

Ejemplos

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- Optional. Flattens all nested columns.
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--Use COALESCE to handle null values.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

Semántica exactly-once

• Configuración del nivel de aislamiento del consumidor

  Todas las aplicaciones que consumen datos de Kafka deben establecer la propiedad isolation.level:

  • read_committed: Lee únicamente los datos confirmados.

  • read_uncommitted (predeterminado): Permite leer datos no confirmados.

  El modo EXACTLY_ONCE depende de read_committed. De lo contrario, los consumidores podrían ver datos no confirmados y romper la consistencia.

• Tiempo de espera de transacción y pérdida de datos

  Durante la recuperación desde un checkpoint, Realtime Compute for Apache Flink considera solo las transacciones confirmadas antes de que dicho checkpoint comenzara. Si el tiempo transcurrido entre la falla de un trabajo y su reinicio supera el tiempo de espera de transacción de Kafka, este cancela automáticamente la transacción abierta, lo que puede provocar pérdida de datos.

  • El valor predeterminado de transaction.max.timeout.ms para un broker de Kafka es de 15 minutos.

  • De forma predeterminada, Flink Kafka Sink establece el parámetro transaction.timeout.ms en 1 hora.

  • Debe aumentar transaction.max.timeout.ms en el broker para que sea mayor o igual que el valor configurado en Flink.

• Pool de productores y checkpoints simultáneos

  El modo EXACTLY_ONCE utiliza un pool de productores de Kafka de tamaño fijo. Cada checkpoint consume un productor de este pool. Si la cantidad de checkpoints simultáneos supera el tamaño del pool, el trabajo falla.

  Configure el tamaño del pool de productores según el número máximo de checkpoints simultáneos.

• Restricciones para reducir el paralelismo

  Si un trabajo falla antes de completar el primer checkpoint, la información original del pool de productores se pierde al reiniciar. Por lo tanto, no reduzca el paralelismo del trabajo antes de que finalice el primer checkpoint. Si la reducción es necesaria, el nuevo paralelismo no debe ser menor que FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Las transacciones bloquean las lecturas

  En el modo read_committed, cualquier transacción que no se haya confirmado ni cancelado bloquea las operaciones de lectura en todo el topic.

  Por ejemplo:

  • La transacción 1 escribe datos.

  • La transacción 2 escribe más datos y se confirma.

  • Mientras la transacción 1 permanezca abierta, los datos de la transacción 2 confirmada resultan invisibles para los consumidores.

  Esto tiene las siguientes implicaciones:

  • Durante la operación normal, la latencia de visibilidad de los datos equivale aproximadamente al intervalo del checkpoint.

  • Si un trabajo falla, cualquier topic donde escribía queda bloqueado para los consumidores hasta que el trabajo se reinicie o la transacción agote su tiempo de espera. En casos extremos, el propio proceso de tiempo de espera de la transacción también afecta las operaciones de lectura.

Preguntas frecuentes

• How do I retrieve JSON data from Kafka using Realtime Compute for Apache Flink?

• Why does an Event Time-based Window produce no output for a ApsaraMQ for Kafka Source Table?

• How do I connect to a secured Kafka Cluster?

• How do I resolve field name conflicts?

• Workspaces and Namespaces