Modo de compatibilidad con Kafka de DataHub

DataHub ahora es compatible con el protocolo de Apache Kafka. Puede usar un SDK estándar de Kafka para conectarse a DataHub y publicar o suscribirse a datos.

Mapeo de conceptos entre DataHub y Kafka

Kafka	DataHub
topic	Project.topic
partition	shard
offset	sequence

Topic de Kafka

DataHub también cuenta con topics. Sin embargo, incluye una capa adicional de recursos llamada proyecto, que no existe en Kafka. Para garantizar la compatibilidad, un topic de Kafka corresponde a la combinación de un proyecto y un topic de DataHub, unidos por un punto (.). Por ejemplo, si tiene un proyecto llamado test_project y un topic llamado test_topic en DataHub, el nombre del topic de Kafka correspondiente es test_project.test_topic.

Partición de Kafka

Una partición de Kafka equivale directamente a un shard de DataHub. Ambos representan una cola ordenada de datos.

Grupo de consumidores de Kafka

Un grupo de consumidores de DataHub funciona de manera similar a uno de Kafka. Puede crear un grupo de consumidores dentro de un proyecto y asociarlo con los topics a los que desea suscribirse. El grupo puede suscribirse a varios topics dentro de ese proyecto. Cuando un grupo de consumidores se vincula a un topic, se crea automáticamente una suscripción que aparece en la página de lista de suscripciones del topic. Si elimina esta suscripción, el grupo no podrá suscribirse al topic y se borrarán todos los offsets de consumo anteriores.

Dado que un grupo de consumidores es un sub-recurso de un proyecto, debe especificarlo junto con el proyecto. Por ejemplo, si su proyecto de DataHub es test_project y su grupo de consumidores es test_group, el nombre del grupo de Kafka es test_project.test_group.

En el cuadro de diálogo Create consumer group, introduzca un Name y una Description, seleccione los topics a los que desea suscribirse en el área Topic, haga clic en el botón > para agregarlos a la lista seleccionada de la derecha y, a continuación, haga clic en Create.

Cada grupo puede suscribirse a un máximo de 50 topics. Si necesita suscribirse a más, envíe un ticket de soporte.

Registro de Kafka

Un registro de Kafka usa un formato clave-valor. Los registros de DataHub están disponibles en dos formatos: Tuple (datos fuertemente estructurados) y Blob (datos binarios). Un registro de Kafka consta de dos partes: un encabezado y un par clave-valor.

Encabezado de Kafka

Un encabezado de Kafka agrega metadatos a los registros, lo cual equivale a un atributo de DataHub. Cuando usa un cliente de Kafka para escribir datos con encabezados, la información del encabezado se almacena como atributos de DataHub. Si el valor de un encabezado de Kafka es NULL, se ignora el encabezado correspondiente. No use "__kafka_key__" como clave de encabezado, ya que es una clave interna reservada.

Datos clave-valor de Kafka

Si un topic de DataHub es de tipo Tuple, su clave corresponde a la primera columna String y su valor corresponde a la segunda columna String.
Si un topic de DataHub es de tipo Blob, su valor corresponde al contenido de los datos y su clave se coloca en un atributo con el formato <"__kafka_key__", key>.

Offset de Kafka

Un offset de Kafka es un entero de 64 bits que comienza en 0 y se autoincrementa dentro de cada partición, lo que otorga a cada registro una posición única. Se usa principalmente para rastrear el progreso del consumidor. Una secuencia de DataHub equivale directamente a un offset de Kafka.

Límites

DataHub no admite transacciones de Kafka, idempotencia, SchemaRegistry ni Log Compaction.

Inicio rápido

Use la versión recomendada del cliente de Kafka 2.4.0. Es compatible con las versiones 0.10.0 a 4.0.
Cree los recursos correspondientes de Project, Topic y Group en DataHub.
Cambie el método de autenticación de seguridad a SASL/SSL, use el mecanismo PLAIN para SASL y configure su par de AccessKey de Alibaba Cloud (AK/SK).
Cambie la información del broker de Kafka al endpoint de DataHub (consulte la lista de dominios de servicio).

Creación de recursos

Primero, inicie sesión en la DataHub console y cree un proyecto.

En el cuadro de diálogo New Project, establezca Name en test_kafka_project y Description en test kafka.

Crear un topic

En la página New Topic de la consola de DataHub, seleccione Direct Create como método de creación, introduzca test_kafka como nombre y seleccione TUPLE como tipo. En la sección Schema details, agregue dos campos llamados key y value, establezca el tipo de ambos en STRING y seleccione Allow NULL. Establezca Shard Count en 1 y Lifecycle en 3, habilite Shard Scaling Mode y deshabilite Multi-version. Introduzca una descripción y haga clic en Create.

Cree un grupo y vincule los topics que necesita consumir. Puede modificar la lista de topics vinculados después de crear el grupo. Si no necesita consumir mensajes, omita este paso.

En el panel New Group, introduzca un Name y una Description, use la lista de transferencia para mover los topics a consumir de la lista disponible de la izquierda a la lista seleccionada de la derecha y, a continuación, haga clic en Create.

Método de autenticación

Cree un archivo llamado kafka_client_producer_jaas.conf y guárdelo en cualquier ruta con el siguiente contenido:

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="accessId"
  password="accessKey";
};

Cree un archivo llamado kafka_client_producer_jaas.conf y guárdelo en cualquier directorio con el siguiente contenido:

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="YOUR_ACCESS_ID"
  password="YOUR_ACCESS_KEY";
};

Ejemplo de productor

Archivo pom de clientes de Kafka

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

Archivo pom.xml del cliente de Kafka:

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

ProducerExample

Ejemplo de consumidor

ConsumerExample

package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample2 {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        kafkaConsumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

ConsumerExample

package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/to/your/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        kafkaConsumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

Ejemplos de Streams

Este código lee datos de input en test_project, convierte las cadenas de clave y valor a minúsculas y reescribe los datos en output.

public class StreamExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.test_kafka_group");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");
        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

Este ejemplo lee datos del topic de entrada en el proyecto test_project, convierte las cadenas de clave y valor a minúsculas y escribe el resultado en el topic de salida.

public class StreamExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/to/your/kafka_client_producer_jaas.conf");
    }
    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.test_kafka_group");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");
        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

Después de iniciar el trabajo de Streams, la asignación de shards tarda aproximadamente un minuto. Luego puede consultar el recuento de tareas en la consola. El número de tareas coincide con el número de shards en el topic de entrada. En este ejemplo, el topic de entrada tiene tres shards.

El resultado de muestreo de salida muestra que los datos se escribieron correctamente. Tras ser procesados por TestMapper en la tarea de Streams, la entrada en mayúsculas se convirtió en una salida en minúsculas: las claves son aaaa, cccc y eeee, y los valores correspondientes son bbbb, dddd y ffff.

Migrar datos de Kafka autoadministrado a DataHub

Cambie la dirección del broker. Para obtener más información, consulte la lista de nombres de dominio a continuación.
Cree recursos en DataHub y modifique los nombres de los recursos en su código. Para obtener más información, consulte la sección de creación de recursos anterior.
Cambie el método de autenticación a SASL_SSL. El mecanismo de autenticación es PLAIN. Establezca username en su AccessKey ID de Alibaba Cloud (AK) y password en su AccessKey Secret (SK).

Apéndice

Descripción general de la configuración

C=Consumidor, P=Productor, S=Streams

Parámetro	C/P/S	Valor recomendado	Obligatorio	Descripción
bootstrap.servers	*	Consulte la lista de nombres de dominio.	Sí
security.protocol	*	SASL_SSL	Sí	Para garantizar una transmisión segura de datos, la escritura de datos desde Kafka hacia DataHub usa cifrado SSL de forma predeterminada.
sasl.mechanism	*	PLAIN	Sí	Autenticación mediante AccessKey. Solo se admite PLAIN.
compression.type	P	LZ4	No	Especifica si se habilita la compresión para la transmisión de datos. Actualmente, solo se admite LZ4.
enable.idempotence	P	false	No	Las versiones 3.0.1 y posteriores de los clientes de Kafka habilitan la idempotencia de forma predeterminada. Dado que DataHub no admite idempotencia, debe deshabilitar manualmente esta función. Esta configuración no es necesaria para clientes anteriores a la versión 3.0.1.
group.id	C	project.group	Sí
partition.assignment.strategy	C	org.apache.kafka.clients.consumer.RangeAssignor	No	El valor predeterminado en Kafka es RangeAssignor. DataHub actualmente solo admite RangeAssignor. No cambie esta configuración.
session.timeout.ms	C/S	[60000, 180000]	No	El valor predeterminado en Kafka es 10000. Sin embargo, DataHub impone un mínimo de 60000, por lo que el valor predeterminado es 60000.
heartbeat.interval.ms	C/S	Dos tercios del valor de session.timeout.ms.	No	El valor predeterminado en Kafka es 3000. Debido a que `session.timeout.ms` tiene un valor predeterminado de 60000, establezca explícitamente este valor en 40000 para evitar solicitudes de latido frecuentes.
application.id	S	project.topic:subId o project.group	Sí	Si usa project.topic:subId, debe coincidir con el topic suscrito. De lo contrario, no se podrán leer los datos. Se recomienda usar project.group.

Lista de nombres de dominio de servicio

Nombre de región	Región	Endpoint público	Endpoint VPC ECS
China East 1 (Hangzhou)	cn-hangzhou	dh-cn-hangzhou.aliyuncs.com:9092	dh-cn-hangzhou-int-vpc.aliyuncs.com:9094
China East 2 (Shanghai)	cn-shanghai	dh-cn-shanghai.aliyuncs.com:9092	dh-cn-shanghai-int-vpc.aliyuncs.com:9094
China North 2 (Beijing)	cn-beijing	dh-cn-beijing.aliyuncs.com:9092	dh-cn-beijing-int-vpc.aliyuncs.com:9094
China (Ulanqab)	cn-wulanchabu	dh-cn-wulanchabu.aliyuncs.com:9092	dh-cn-wulanchabu-int-vpc.aliyuncs.com:9094
China South 1 (Shenzhen)	cn-shenzhen	dh-cn-shenzhen.aliyuncs.com:9092	dh-cn-shenzhen-int-vpc.aliyuncs.com:9094
China North 3 (Zhangjiakou)	cn-zhangjiakou	dh-cn-zhangjiakou.aliyuncs.com:9092	dh-cn-zhangjiakou-int-vpc.aliyuncs.com:9094
Asia Pacific SE 1 (Singapore)	ap-southeast-1	dh-ap-southeast-1.aliyuncs.com:9092	dh-ap-southeast-1-int-vpc.aliyuncs.com:9094
Asia Pacific SE 3 (Kuala Lumpur)	ap-southeast-3	dh-ap-southeast-3.aliyuncs.com:9092	dh-ap-southeast-3-int-vpc.aliyuncs.com:9094
Europe Central 1 (Frankfurt)	eu-central-1	dh-eu-central-1.aliyuncs.com:9092	dh-eu-central-1-int-vpc.aliyuncs.com:9094
China East 2 (Shanghai) Finance	cn-shanghai-finance-1	dh-cn-shanghai-finance-1.aliyuncs.com:9092	dh-cn-shanghai-finance-1-int-vpc.aliyuncs.com:9094
China (Hong Kong)	cn-hongkong	dh-cn-hongkong.aliyuncs.com:9092	dh-cn-hongkong-int-vpc.aliyuncs.com:9094

Compatibilidad con la API de Kafka

La official Kafka documentation enumera todas las API disponibles. La siguiente tabla muestra las API que DataHub admite en el modo de compatibilidad con Kafka.

API	Descripción
Produce	Escribe datos.
Fetch	Lee datos.
ListOffsets	Obtiene offsets basados en tiempo. Las versiones anteriores devuelven una lista de offsets, mientras que las versiones posteriores devuelven un único offset.
Metadata	Obtiene metadatos para operaciones de lectura y escritura.
OffsetCommit	Confirma un offset.
OffsetFetch	Obtiene un offset.
FindCoordinator	Encuentra el broker que aloja al coordinador y devuelve su dirección IP virtual (VIP).
JoinGroup	Se une a un grupo.
Heartbeat	Envía un latido.
LeaveGroup	Abandona un grupo.
SyncGroup	Lo usa el líder del grupo para enviar el plan de asignación de particiones y todos los miembros para recibirlo.
SaslHandshake	Gestiona el protocolo de enlace Simple Authentication and Security Layer (SASL) para la autenticación.
ApiVersions	Obtiene todas las API disponibles.
CreateTopics	Crea un topic.
DeleteTopics	Elimina un topic.
OffsetForLeaderEpoch	Obtiene el offset más reciente.
SaslAuthenticate	Autentica al cliente mediante SASL.
CreatePartitions	Agrega una partición.
DeleteGroups	Elimina un grupo.
OffsetDelete	Elimina un offset.