Plug-in de Canal - DataHub - Centro de documentación de Alibaba Cloud

Canal lee los registros binarios (binlogs) de MySQL para capturar cambios incrementales y transmitirlos a temas de DataHub mediante el protocolo Kafka.

Funcionamiento de Canal con DataHub

Nota

Canal admite MySQL 5.1.x, 5.5.x, 5.6.x, 5.7.x y 8.0.x como bases de datos de origen.

DataHub es compatible con el protocolo Kafka, por lo que Canal puede escribir datos incrementales de MySQL directamente en temas de DataHub. El paquete de Canal personalizado para DataHub incluye dos cambios respecto a la versión de código abierto:

Se eliminó la lógica que reemplaza los puntos (.) por guiones bajos (_) en los nombres de temas de Kafka. Esto permite que Canal asigne el nombre del tema de Kafka al tema correcto de DataHub mediante el formato ProjectName.TopicName.
Se agregó la variable de entorno -Djava.security.auth.login.config=$kafka_jaas_conf al script de inicio para admitir la autenticación PLAIN Simple Authentication and Security Layer (SASL).

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Una base de datos MySQL que ejecute la versión 5.1.x, 5.5.x, 5.6.x, 5.7.x u 8.0.x. Consulte QuickStart para obtener detalles sobre la configuración.
Un tema de DataHub de tipo TUPLE creado en el proyecto de destino. Para conocer los requisitos del tema, consulte Kafka compatibility.

Para consultar todos los parámetros de configuración de Canal y las opciones avanzadas, visite Canal.

Configuración de Canal para transmitir datos de MySQL a DataHub

Paso 1: Descargue el paquete de Canal

Descargue el paquete de Canal personalizado para DataHub: canal.deployer-1.1.5-SNAPSHOT.tar.gz

Utilice este paquete en lugar de la versión estándar de código abierto de Canal. Es posible que una versión de Canal no modificada para DataHub no pueda escribir datos en dicho servicio.

Paso 2: Extraiga el paquete

mkdir -p /usr/local/canal
tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz -C /usr/local/canal

Paso 3: Configure Canal

Edite tres archivos de configuración antes de iniciar Canal.

**3.1 Configuración de la instancia: conf/example/instance.properties**

# Modify the database information as needed.
#################################################
...
canal.instance.master.address=192.168.1.20:3306
# username/password: the username and password of the database.
...
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
...
# mq config
canal.mq.topic=test_project.test_topic
# Specify a dynamic topic based on the database name or table name.
#canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
# Database name.Table name: the unique primary key. Multiple tables are separated with comma (,).
#canal.mq.partitionHash=mytest.person:id,mytest.role:id
#################################################

Establezca canal.mq.topic con el valor <ProjectName>.<TopicName>; el formato separado por puntos se asigna directamente a un tema de DataHub. Para configurar el enrutamiento dinámico de temas según los nombres de bases de datos o tablas, así como las claves hash de partición, consulte MQ-related parameters.

**3.2 Configuración de Canal: conf/canal.properties**

# ...
canal.serverMode = kafka
# ...
kafka.bootstrap.servers = dh-cn-hangzhou.aliyuncs.com:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

kafka.security.protocol = SASL_SSL
kafka.sasl.mechanism = PLAIN

Los siguientes parámetros son obligatorios:

Parámetro	Valor	Descripción
`canal.serverMode`	`kafka`	Indica a Canal que utilice el protocolo de productor de Kafka
`kafka.bootstrap.servers`	Endpoint de DataHub	El endpoint de DataHub en la región donde reside el tema de destino. Para consultar los endpoints disponibles, consulte Kafka compatibility.
`kafka.security.protocol`	`SASL_SSL`	Obligatorio para la autenticación de DataHub
`kafka.sasl.mechanism`	`PLAIN`	Obligatorio para la autenticación SASL/PLAIN de DataHub

El resto de los parámetros son opcionales y se pueden ajustar según sus requisitos de throughput.

**3.3 Configuración de Java Authentication and Authorization Service (JAAS): conf/kafka_client_producer_jaas.conf**

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

Paso 4: Inicie Canal y compruebe el flujo de datos

Inicie Canal

cd /usr/local/canal/
sh bin/startup.sh

Compruebe que Canal se esté ejecutando

Revise el registro principal de Canal para confirmar que el servidor se inició correctamente:

2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]
2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

Ejecute vi logs/canal/canal.log para consultar el registro principal. Ejecute vi logs/example/example.log para revisar el registro de la instancia:

2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....

Verifique la captura de datos

Ejecute vi logs/example/meta.log para confirmar que Canal está capturando los cambios en la base de datos. Se genera un registro en el archivo meta.log por cada inserción, eliminación y modificación en la base de datos. Consulte este archivo para comprobar si Canal ha recopilado datos.

tail -f example/meta.log
2020-07-29 09:21:05.110 - clientId:1001 cursor:[log.000001,29723,1591190230000,1,] address[/127.0.0.1:3306]
2020-07-29 09:23:46.109 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[localhost/127.0.0.1:3306]
2020-07-29 09:24:50.547 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[/127.0.0.1:3306]
2020-07-29 09:26:45.547 - clientId:1001 cursor:[log.000001,30143,1595986005000,1,] address[localhost/127.0.0.1:3306]
2020-07-29 09:30:04.546 - clientId:1001 cursor:[log.000001,30467,1595986204000,1,] address[localhost/127.0.0.1:3306]
2020-07-29 09:30:16.546 - clientId:1001 cursor:[log.000001,30734,1595986215000,1,] address[localhost/127.0.0.1:3306]
2020-07-29 09:30:36.547 - clientId:1001 cursor:[log.000001,31001,1595986236000,1,] address[localhost/127.0.0.1:3306]

Detenga Canal

cd /usr/local/canal/
sh bin/stop.sh

Ejemplo

El siguiente ejemplo muestra cómo los cambios de fila en MySQL aparecen como registros de DataHub después de que Canal los procesa.

Esquema del tema de DataHub

El tema de destino es de tipo TUPLE y cuenta con el siguiente esquema:

+-------+------+----------+-------------+
| Index | name |   type   |  allow NULL |
+-------+------+----------+-------------+
|   0   |  key |  STRING  |     true    |
|   1   |  val |  STRING  |     true    |
+-------+------+----------+-------------+

Tabla de MySQL de origen

mysql> desc orders;
+-------+---------+------+-----+---------+-------+
| Field | Type    | Null | Key | Default | Extra |
+-------+---------+------+-----+---------+-------+
| oid   | int(11) | YES  |     | NULL    |       |
| pid   | int(11) | YES  |     | NULL    |       |
| num   | int(11) | YES  |     | NULL    |       |
+-------+---------+------+-----+---------+-------+
3 rows in set (0.00 sec)

Registro resultante en DataHub

Una vez que se ejecuta mysql> insert into orders values(1,2,3);, Canal escribe el cambio en DataHub. El campo key es nulo y el campo val contiene el evento de cambio completo como una cadena JSON:

{
    "data":[
        {
            "oid":"1",
            "pid":"2",
            "num":"3"
        }
    ],
    "database":"ggtt",
    "es":1591092305000,
    "id":2,
    "isDdl":false,
    "mysqlType":{
        "oid":"int(11)",
        "pid":"int(11)",
        "num":"int(11)"
    },
    "old":null,
    "pkNames":null,
    "sql":"",
    "sqlType":{
        "oid":4,
        "pid":4,
        "num":4
    },
    "table":"orders",
    "ts":1591092305813,
    "type":"INSERT"
}

Pasos siguientes

Una vez que Canal esté en ejecución y aparezcan datos en meta.log, diríjase a la consola de DataHub y revise el tema de destino para confirmar que los registros están llegando. A partir de ahí, conecte un consumidor, como un trabajo de Flink o una aplicación posterior, para procesar los datos incrementales.