How to use the flink-dts-connector - Data Transmission Service - Centro de documentación de Alibaba Cloud

Una vez que configure una instancia de suscripción, puede usar el flink-dts-connector para consumir datos de la instancia mediante un cliente Flink. En este tema se describe cómo utilizar el flink-dts-connector.

Notas de uso

Este conector es compatible con clientes Flink que usan la API DataStream, Table API o SQL.
Si su cliente Flink usa Table API o SQL, solo puede consumir datos de una tabla por cada configuración. Para consumir datos de varias tablas, debe configurar y ejecutar varias tareas independientes.

Procedimiento

En este tema se utiliza IntelliJ IDEA Community Edition 2020.1 para Windows con el fin de demostrar cómo usar el flink-dts-connector para consumir datos de una instancia de suscripción.

Cree una tarea de seguimiento de cambios. Para obtener más información, consulte los temas correspondientes en Change tracking overview.
Cree uno o más grupos de consumidores. Para obtener más información, consulte Add a consumer group.
Descargue y descomprima el flink-dts-connector.
Abra IntelliJ IDEA y haga clic en Open or Import.
En el cuadro de diálogo que aparece, vaya al directorio descomprimido de flink-dts-connector, expanda las carpetas y busque el archivo Project Object Model (pom.xml).
En el cuadro de diálogo que aparece, seleccione Open as Project.

Agregue la siguiente dependencia al archivo pom.xml:

<dependency>
      <groupId>com.alibaba.flink</groupId>
      <artifactId>flink-dts-connector</artifactId>
      <version>1.1.1-SNAPSHOT</version>
      <classifier>jar-with-dependencies</classifier>
</dependency>

En IntelliJ IDEA, expanda las carpetas y seleccione el archivo Java que corresponde al tipo de programa de su conector Flink.
- Si su cliente Flink usa la API DataStream, haga doble clic en el archivo flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java y siga estos pasos:
  1. En la parte superior de la ventana de IntelliJ IDEA, haga clic en el ícono Run.
  2. En el menú que aparece, haga clic en DtsExample > Edit.
  3. En el campo Program arguments, introduzca los parámetros y sus valores según el siguiente ejemplo. A continuación, haga clic en Run para iniciar el flink-dts-connector.
    
    Nota
    Para obtener más información sobre los parámetros y cómo obtener sus valores, consulte Parameters.
    --broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043
  4. La salida confirma que el cliente puede suscribirse a los cambios de datos de la base de datos de origen.
    
    Nota
    Para consultar registros específicos de cambios de datos, inicie sesión en la interfaz de Task Manager de su cliente Flink.
- Si su cliente Flink usa Table API o SQL, haga doble clic en el archivo flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java y siga estos pasos:
  
  Nota
  Cada archivo DtsTableISelectTCaseTest.java se puede configurar para consumir datos de una sola tabla. Para configurar y consumir datos de varias tablas, debe repetir la configuración y ejecutar varias tareas independientes.
  1. Agregue dos barras diagonales (//) para comentar la línea de código especificada.
  2. Configure la información de la única tabla de la que desea consumir datos. Se admiten instrucciones SQL.
  3. Configure los parámetros de la instancia de suscripción. Para obtener más información sobre los parámetros y cómo obtener sus valores, consulte Parameters.
  4. En la parte superior de la ventana de IntelliJ IDEA, haga clic en Run'DtsTableISelectTCaseTest' para iniciar el flink-dts-connector.
  5. La salida confirma que el cliente puede suscribirse a los cambios de datos de la base de datos de origen.
    
    Nota
    Para consultar registros específicos de cambios de datos, inicie sesión en la interfaz de Task Manager de su cliente Flink.

Parámetros

DtsExample.java	DtsTableISelectTCaseTest.java	Descripción	Método de consulta
`broker-url`	`dts.server`	El endpoint de red y el puerto de la instancia de suscripción. Nota Si la instancia ECS donde se despliega su cliente Flink y la instancia de suscripción están en la misma red clásica o Virtual Private Cloud (VPC), se recomienda usar un endpoint interno para la suscripción de datos. De esta forma se minimiza la latencia de red. No se recomienda usar un endpoint público debido a la posible inestabilidad de la red.	En la consola DTS, haga clic en el ID de la instancia de suscripción de destino. En la página Basic Information, puede obtener la información de Topic y Network.
`topic`	`topic`	El tema de la instancia de suscripción.
`sid`	`dts.sid`	El ID del grupo de consumidores.	En la consola DTS, haga clic en el ID de la instancia de suscripción de destino. En el panel de navegación izquierdo, haga clic en Consume Data. Puede obtener el valor de Consumer Group ID/Name y la Account del grupo de consumidores. Nota La contraseña de la cuenta del grupo de consumidores se especifica cuando se crea dicho grupo.
`user`	`dts.user`	La cuenta del grupo de consumidores. Aviso Si no utiliza el flink-dts-connector proporcionado en este tema, debe establecer el nombre de usuario con el formato `<consumer_group_account>-<consumer_group_ID>`, como `dtstest-dtsaebpv`. De lo contrario, la conexión fallará.
`password`	`dts.password`	La contraseña de la cuenta.
`checkpoint`	`dts.checkpoint`	El desplazamiento del consumidor. Este parámetro especifica la marca de tiempo Unix del primer registro de datos que se va a consumir. Ejemplo: 1624440043. Nota Puede usar el desplazamiento del consumidor para: Reanudar el consumo desde el último desplazamiento después de una interrupción de la aplicación y así evitar la pérdida de datos. Especificar un desplazamiento para comenzar a consumir datos desde un momento determinado.	La hora de inicio del consumo debe estar dentro del rango de datos de la instancia de suscripción y convertirse a una marca de tiempo Unix. Nota Puede ver el rango de datos de la instancia de suscripción en la columna Data Range de la lista de tareas de suscripción. Puede usar un motor de búsqueda para encontrar una herramienta de conversión de marcas de tiempo Unix.
N/D	`dts-cdc.table.name`	El objeto suscrito. Solo puede especificar una tabla. Los requisitos de formato son los siguientes: Si el valor de Database Type es MySQL, PolarDB for MySQL, PolarDB-X 1.0 o PolarDB-X 2.0, use el formato `<database_name>.<table_name>`. Para otros tipos de bases de datos, use el formato `<schema_name>.<table_name>`.	En la consola DTS, haga clic en el ID de la instancia de suscripción de destino. En la parte superior de la página Basic Information o Task Management, haga clic en View Objects para buscar la base de datos y la tabla suscritas.

Preguntas frecuentes

Mensaje de error	Causa posible	Solución
`Cluster changed from * to *, consumer require restart.`	Un cambio en el módulo DStore, que Data Transmission Service (DTS) utiliza para leer datos incrementales, provoca que se pierda el desplazamiento del consumidor del cliente Flink.	No reinicie el cliente. En su lugar, consulte el desplazamiento del consumidor del cliente y vuelva a pasar el valor al parámetro `checkpoint` o `dts.checkpoint` en el archivo DtsExample.java o DtsTableISelectTCaseTest.java para reanudar el consumo de datos.