Data Transmission Service:Uso de un cliente Kafka para consumir datos rastreados

Notas de uso

• Si habilita el commit automático al usar la función de seguimiento de cambios, es posible que algunos datos se confirmen antes de consumirse, lo cual provoca pérdida de información. Se recomienda realizar el commit de los datos manualmente.

  Nota

  Si falla el commit de los datos, reinicie el cliente para continuar consumiéndolos desde el último punto de control de consumo registrado. No obstante, durante este periodo podrían generarse datos duplicados que deberá filtrar manualmente.

• Los datos se serializan y almacenan en formato Avro. Para obtener más información, consulte Record.avsc.

  Aviso

  Si no utiliza el cliente Kafka descrito en este tema, debe analizar sintácticamente los datos rastreados según el esquema Avro (Ejemplo de deserialización Avro de DTS) y comprobar los datos analizados.

• La unidad de búsqueda es el segundo cuando Data Transmission Service (DTS) invoca la operación offsetForTimes. En cambio, la unidad es el milisegundo cuando un cliente Kafka nativo ejecuta esta misma operación.

• Pueden producirse conexiones transitorias entre un cliente Kafka y el servidor de seguimiento de cambios debido a diversos motivos, como la recuperación ante desastres. Si no emplea el cliente Kafka descrito en este tema, asegúrese de que su cliente cuente con capacidades de reconexión de red.

• Al utilizar un cliente Kafka nativo para consumir datos rastreados, el módulo de recolección de datos incrementales podría modificarse en DTS. En el modo de suscripción, se elimina el punto de control de consumo que el cliente Kafka guarda en el servidor DTS. Por ello, debe especificar un punto de control de consumo acorde a sus necesidades empresariales. Si desea consumir datos en modo de suscripción, se recomienda usar la demostración del SDK proporcionada por DTS para rastrear y consumir los datos, o bien gestionar manualmente el punto de control de consumo. Para más detalles, consulte Consume subscribed data using an SDK y la sección Manage the consumption checkpoint de este tema.

Ejecución del cliente Kafka

Descargue la demostración del cliente Kafka. Para obtener más información sobre cómo usar la demostración, consulte Readme.

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

Procedimiento

A continuación se muestran los pasos para ejecutar el cliente Kafka y consumir datos rastreados. En este ejemplo se utiliza IntelliJ IDEA Community Edition 2018.1.4 para Windows.

1. Cree una instancia de seguimiento de cambios. Para más información, consulte Change tracking overview.

2. Cree uno o más grupos de consumidores. Para más información, consulte Add a consumer group.

3. Descargue el paquete de la demostración del cliente Kafka y descomprímalo.

   Nota

   Haga clic en y seleccione Download ZIP para descargar el paquete.

4. Abra IntelliJ IDEA. En la ventana que aparece, haga clic en Open.

5. En el cuadro de diálogo que aparece, vaya al directorio donde reside la demostración descargada. Localice el archivo pom.xml.

   Vaya a kafkademo > subscribe_example-master > javaimpl, seleccione pom.xml y luego haga clic en OK.

6. En el cuadro de diálogo que aparece, seleccione Open as Project.

7. En la ventana de herramientas Project de IntelliJ IDEA, navegue por las carpetas para encontrar el archivo de demostración del cliente Kafka y haga doble clic en él. El nombre del archivo es NotifyDemoDB.java.

8. Especifique los parámetros en el archivo NotifyDemoDB.java.

   public static Properties getConfigs() {
       Properties properties = new Properties();
       // user password and sid for auth
       properties.setProperty(USER_NAME, "dtstest");
       properties.setProperty(PASSWORD_NAME, "xxx");
       properties.setProperty(SID_NAME, "dtsxxx");
       // kafka consumer group general same with sid
       properties.setProperty(GROUP_NAME, "dtsxxx");
       // topic to consume, partition is 0
       properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_xxx");
       // kafka broker url
       properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-xxx.com:18001");
       // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
       properties.setProperty(INITIAL_CHECKPOINT_NAME, "1583307907");
       // if force use config checkpoint when start. for checkpoint reset
       properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "true");
       // use consumer assign or subscribe interface
       // when use subscribe mode, group config is required. kafka consumer group is enabled
       properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
       return properties;
   }

   Parámetro	Descripción	Método para obtener el valor del parámetro
   USER_NAME	Nombre de usuario de la cuenta del grupo de consumidores. Aviso Si no utiliza el cliente Kafka descrito en este tema, debe especificar este parámetro con el siguiente formato: <Username>-<Consumer group ID>. Ejemplo: dtstest-dtsaebpv. De lo contrario, la conexión fallará.	En la consola de DTS, localice la instancia de seguimiento de cambios que desea gestionar y haga clic en su ID. En el panel de navegación izquierdo, haga clic en Consume Data. En la página que aparece podrá consultar la información del grupo de consumidores, como su ID, nombre y cuenta. Nota La contraseña de la cuenta del grupo de consumidores se establece al crear dicho grupo.
   PASSWORD_NAME	Contraseña de la cuenta.
   SID_NAME	ID del grupo de consumidores.
   GROUP_NAME	Nombre del grupo de consumidores. Establezca este parámetro con el ID del grupo de consumidores.
   KAFKA_TOPIC	Nombre del tema rastreado de la instancia de seguimiento de cambios.	En la consola de DTS, localice la instancia de seguimiento de cambios que desea gestionar y haga clic en su ID. En la página Basic Information puede consultar la información del tema y de la red. En la sección Basic Information de la página de detalles de la tarea de seguimiento de cambios, obtenga el valor de Subscription Topic. En la sección Network, obtenga la VPC network address (ejemplo de formato: xxx.aliyuncs.com:18003).
   KAFKA_BROKER_URL_NAME	Endpoint de la instancia de seguimiento de cambios. Nota Si rastrea cambios de datos a través de redes internas, la latencia de red es mínima. Esto aplica cuando la instancia de Elastic Compute Service (ECS) donde despliega el cliente Kafka se encuentra en la red clásica o en la misma nube privada virtual (VPC) que la instancia de seguimiento de cambios.
   INITIAL_CHECKPOINT_NAME	Punto de control de consumo de los datos consumidos. El valor es una marca de tiempo UNIX. Ejemplo: 1592269238. Nota Debe guardar el punto de control de consumo por las siguientes razones: Si se interrumpe el proceso de consumo, puede especificar el punto de control en el cliente Kafka para reanudar el consumo de datos y evitar su pérdida. Al iniciar el cliente Kafka, puede especificar el punto de control de consumo para procesar los datos según sus necesidades empresariales. Si el parámetro SUBSCRIBE_MODE_NAME está configurado como subscribe, el parámetro INITIAL_CHECKPOINT_NAME que especifique solo surtirá efecto la primera vez que inicie el cliente Kafka.	El punto de control de consumo de los datos consumidos debe encontrarse dentro del rango de datos de la instancia de seguimiento de cambios. Este punto de control debe convertirse a una marca de tiempo UNIX. En la lista de tareas de seguimiento de cambios de DTS, puede consultar el campo Data Range de la tarea correspondiente, el cual muestra la hora de inicio y fin de los datos disponibles para consumo. Utilice este campo para determinar el rango de valores válidos de INITIAL_CHECKPOINT_NAME. Nota Puede consultar el rango de datos de la instancia de seguimiento de cambios en la columna Data Range de la página Change Tracking Tasks. Puede utilizar un motor de búsqueda para encontrar un convertidor de marcas de tiempo UNIX.
   USE_CONFIG_CHECKPOINT_NAME	Especifica si se debe forzar al cliente a consumir datos desde el punto de control de consumo indicado. Valor predeterminado: true. Puede establecer este parámetro en true para evitar la pérdida de datos recibidos pero aún no procesados.	Ninguna
   SUBSCRIBE_MODE_NAME	Especifica si se ejecutarán dos o más clientes Kafka para un mismo grupo de consumidores. Si desea utilizar esta función, establezca este parámetro en subscribe para dichos clientes Kafka. El valor predeterminado es assign, lo que indica que la función no se utiliza. Se recomienda desplegar solo un cliente Kafka por grupo de consumidores.	Ninguna

9. En la barra de menú superior de IntelliJ IDEA, vaya a Run > Run para ejecutar el cliente.

   Nota

   Si ejecuta IntelliJ IDEA por primera vez, se requerirá cierto tiempo para cargar e instalar las dependencias correspondientes.

Resultados en el cliente Kafka

El siguiente resultado muestra que el cliente Kafka puede rastrear cambios de datos desde la base de datos de origen.

[2020-03-09 10:41:52,408] INFO [Consumer clientId=consumer-1, groupId=dts_xxx] Discovered coordinator xxx (id: xxx rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-03-09 10:41:57,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721711, offset: 1732521, info: 1583721711] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:41:57,571] INFO EtlRecordProcessor: haven't receive records from generator for  5s (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:02,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721721, offset: 1732539, info: 1583721721] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:07,204] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721726, offset: 1732544, info: 1583721726] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:12,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721731, offset: 1732548, info: 1583721731] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:17,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721736, offset: 1732554, info: 1583721736] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:22,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721741, offset: 1732559, info: 1583721741] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:27,206] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721746, offset: 1732569, info: 1583721746] (recordprocessor.EtlRecordProcessor)

Puede eliminar las barras diagonales dobles (//) de la cadena //log.info(ret) en la línea 25 del archivo NotifyDemoDB.java. A continuación, ejecute nuevamente el cliente para consultar la información de cambios de datos.

Preguntas frecuentes

• P: ¿Por qué necesito registrar el punto de control de consumo del cliente Kafka?

  R: El punto de control de consumo registrado por DTS corresponde al momento en que DTS recibe una operación de commit del cliente Kafka. Este punto registrado puede diferir del tiempo real de consumo. Si una aplicación empresarial o el cliente Kafka se interrumpen inesperadamente, podrá especificar un punto de control de consumo preciso para continuar el procesamiento de datos, evitando así la pérdida de información o el consumo duplicado.

Gestión del punto de control de consumo

1. Configure el cliente Kafka para que escuche el cambio del módulo de recolección de datos en DTS.

   Puede configurar las propiedades del consumidor del cliente Kafka para escuchar el cambio del módulo de recolección de datos en DTS. El siguiente código muestra un ejemplo de cómo configurar dichas propiedades:

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   El siguiente código proporciona un ejemplo de cómo implementar ClusterSwitchListener:

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. Especifique el punto de control de consumo basándose en el cambio detectado del módulo de recolección de datos en DTS.

   Establezca el punto de control de inicio del próximo seguimiento de datos con la marca de tiempo de la última entrada de datos rastreada que consumió el cliente. El siguiente código muestra un ejemplo de cómo especificar el punto de control de consumo:

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the consumption checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   Nota

   Para obtener más información sobre los ejemplos, consulte KafkaRecordFetcher.

Asignaciones entre tipos de datos MySQL y valores dataTypeNumber

Para obtener más información, consulte Campo SQL Type.

Asignaciones entre tipos de datos Oracle y valores dataTypeNumber

Asignaciones entre tipos de datos PostgreSQL y valores dataTypeNumber