Realtime Compute for Apache Flink:Ingesta de registros en tiempo real

Requisitos previos

• Asegúrese de que su usuario o rol de RAM cuente con los permisos obligatorios para acceder a la consola de Realtime Compute for Apache Flink. Para obtener más información, consulte Permissions.

• Ya creó un espacio de trabajo de Flink. Para obtener más información, consulte Create a workspace.

• Almacenamiento ascendente y descendente

  • Ya creó una instancia de ApsaraMQ for Kafka. Para obtener más información, consulte Step 2: Purchase and deploy an instance.

  • Ya creó una instancia de Hologres. Para obtener más información, consulte Purchase a Hologres instance.

  Nota

  Sus instancias de ApsaraMQ for Kafka y Hologres deben estar en la misma región y VPC que su espacio de trabajo de Realtime Compute for Apache Flink. De lo contrario, debe establecer la conectividad de red. Para obtener más información, consulte Access services across VPCs o Access the internet.

Paso 1: Configurar una lista blanca de IP

Para permitir que Flink acceda a sus instancias de Kafka y Hologres, agregue el bloque CIDR de su espacio de trabajo de Flink a las listas blancas de IP de ambos servicios.

1. Obtenga el bloque CIDR de VPC de su espacio de trabajo de Flink.

   1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

   2. En la columna Actions del workspace de destino, elija More > Workspace Details.

   3. En el cuadro de diálogo Workspace Details, busque el CIDR block del VSwitch.

      El cuadro de diálogo muestra la información básica del espacio de trabajo y una lista de VSwitches. En la columna CIDR block, consulte el bloque CIDR de cada zona de disponibilidad. Tome nota de este bloque CIDR para el siguiente paso.

2. Agregue el bloque CIDR del espacio de trabajo de Flink a la lista blanca de IP de su instancia de Kafka.

   Debe configurar la lista blanca de IP para un endpoint de VPC. Para conocer los pasos detallados, consulte Configure an IP whitelist. En el cuadro de diálogo de edición de la lista blanca, haga clic en Add Whitelist IP para agregar el bloque CIDR.

3. Agregue el bloque CIDR del espacio de trabajo de Flink a la lista blanca de IP de su instancia de Hologres.

   Inicie sesión en su instancia de Hologres y configure su lista blanca de IP. Para conocer los pasos detallados, consulte IP whitelist. En la página de configuración de la lista blanca del Security Center de HoloWeb, introduzca el bloque CIDR en el campo IP Address del cuadro de diálogo Edit IP Whitelist y haga clic en OK.

Paso 2: Preparar los datos de prueba de Kafka

Utilice el conector Faker de Realtime Compute for Apache Flink para generar datos y escribirlos en ApsaraMQ for Kafka. Siga estos pasos en la consola de desarrollo de Realtime Compute for Apache Flink.

1. En la consola de ApsaraMQ for Kafka, cree un topic llamado users.

   Para obtener más información, consulte Create a topic.

2. Cree un trabajo para escribir datos en ApsaraMQ for Kafka.

   1. Inicie sesión en la consola de desarrollo de Realtime Compute for Apache Flink.

   2. En la columna Actions del espacio de trabajo de destino, haga clic en Console.

   3. En el panel de navegación izquierdo, elija Development > ETL.

   4. Haga clic en el icono y luego en New Blank Stream Draft. Introduzca un Name y seleccione una Engine Version.

      Realtime Compute for Apache Flink también ofrece diversas plantillas de código y de sincronización de datos. Cada una incluye casos de uso específicos, ejemplos de código e instrucciones. Haga clic en una plantilla para conocer rápidamente las funciones y la sintaxis de Realtime Compute for Apache Flink e implementar su lógica de negocio. Para obtener más información, consulte Code templates y Data synchronization templates.

      Parámetro	Descripción	Ejemplo
      Name	El nombre del trabajo. Nota El nombre del trabajo debe ser único dentro del espacio de trabajo actual.	flink-test
      Engine Version	La versión del motor de Flink para el trabajo actual. Seleccione una versión etiquetada como Recommended o Stable para obtener mayor confiabilidad y rendimiento. Para obtener más información sobre las versiones del motor, consulte Release notes y Engine versions.	vvr-8.0.8-flink-1.17

   5. Haga clic en Create.

   6. Escriba las sentencias SQL del trabajo.

      Copie el siguiente código en el editor y modifique los parámetros según su entorno.

      CREATE TEMPORARY TABLE source (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        event_time TIMESTAMP
      ) WITH (
        'connector' = 'faker',
        'number-of-rows' = '100',
        'rows-per-second' = '10',
        'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}',
        'fields.first_name.expression' = '#{name.firstName}',
        'fields.last_name.expression' = '#{name.lastName}',
        'fields.address.country.expression' = '#{Address.country}',
        'fields.address.state.expression' = '#{Address.state}',
        'fields.address.city.expression' = '#{Address.city}',
        'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}'
      );
      
      CREATE TEMPORARY TABLE sink (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        `timestamp` TIMESTAMP METADATA
      ) WITH (
        'connector' = 'kafka',
        'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092',
        'topic' = 'users',
        'format' = 'json',
        'properties.enable.idempotence'='false'
      );
      
      INSERT INTO sink SELECT * FROM source;

      A continuación se describen los parámetros que debe modificar.

      Parámetro	Valor de ejemplo	Descripción
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	Los endpoints de los brokers de ApsaraMQ for Kafka. Una lista separada por comas de endpoints de brokers con el formato 'host:port'. Puede encontrar el endpoint de Domain Name de VPC en la sección Endpoint Information de la página Instance Details.
      topic	users	El nombre del topic de ApsaraMQ for Kafka.

3. Inicie el trabajo.

   1. En la página Development > ETL, haga clic en Deploy.

   2. En el cuadro de diálogo Deploy draft, haga clic en Confirm.

   3. Configure los recursos del trabajo. Para obtener más información, consulte Configure resources for a job.

   4. En la página O&M > Deployments, busque el despliegue de destino y haga clic en Start en la columna Actions. Para obtener más información sobre la configuración de inicio, consulte Start a deployment.

   5. En la página Deployments, puede supervisar la información de ejecución y el estado del despliegue.

      Dado que el origen Faker genera un flujo acotado, el estado del despliegue cambia a FINISHED aproximadamente un minuto después de iniciarse. Cuando el despliegue finaliza, los datos ya se escribieron en el topic 'users'. El siguiente código muestra un ejemplo de los datos con formato JSON.

      {
        "id": 765,
        "first_name": "Barry",
        "last_name": "Pollich",
        "address": {
          "country": "United Arab Emirates",
          "state": "Nevada",
          "city": "Powlowskifurt"
        }
      }

Paso 3: Crear e iniciar un trabajo de sincronización de datos

Paso 4: Consultar el resultado de la sincronización completa

1. Inicie sesión en la consola de administración de Hologres.

2. En la página Instances, haga clic en el nombre de su instancia de destino.

3. En la esquina superior derecha de la página, haga clic en Connect to Instance.

4. En la pestaña Metadata Management, consulte el esquema de tabla y los datos de la tabla users sincronizada en la base de datos flink_test_db.

   En el árbol de navegación izquierdo, expanda el nombre de su instancia > flink_test_db > test_schema > Tables. Aparecerá la tabla users sincronizada.

   El esquema y los datos de la tabla sincronizada son los siguientes.

   • Esquema de la tabla

     Haga doble clic en el nombre de la tabla users para consultar su esquema.

     El esquema de la tabla users contiene los siguientes campos: id (BIGINT, clave principal), first_name (TEXT), last_name (TEXT), address.country (TEXT), address.state (TEXT) y address.city (TEXT).

     Nota

     Durante la sincronización completa de datos, se recomienda definir los metadatos partition y offset de Kafka como clave principal de la tabla de Hologres. Esto evita la duplicación de datos si el trabajo presenta un failover y retransmite información.

   • Datos de la tabla

     En la esquina superior derecha de la página de la tabla users, haga clic en Query table. Introduzca el siguiente comando y haga clic en Run.

     SELECT * FROM test_schema.users;

     El comando devuelve el siguiente resultado.

     La consulta devuelve varias filas, lo cual confirma que los registros se sincronizaron correctamente en la tabla users. Cada fila devuelta contiene datos completos para las columnas id, first_name, last_name, address.country, address.state y address.city.

Paso 5: Observar la sincronización automática del esquema

1. En la consola de ApsaraMQ for Kafka, envíe manualmente un mensaje que contenga una columna nueva.

   1. Inicie sesión en la consola de ApsaraMQ for Kafka.

   2. En la página Instances, haga clic en el nombre de la instancia de destino.

   3. En la página Topics, haga clic en el nombre del topic de destino (users).

   4. Haga clic en Send Message.

   5. Configure el mensaje.

      En el cuadro de diálogo Start to Send and Consume Message, configure los parámetros de la siguiente manera.

      Parámetro	Ejemplo
      Method of Sending	Seleccione Console.
      Message Key	Introduzca flinktest.
      Message Content	Copie y pegue a continuación el contenido JSON en el campo Message Content. { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } } Nota En este ejemplo, house-points es una columna anidada nueva.
      Send to Specified Partition	Seleccione Yes.
      Partition ID	Introduzca 0.

   6. Haga clic en OK.

2. En la consola de Hologres, consulte los cambios de esquema y de datos en la tabla users.

   1. Inicie sesión en la consola de Hologres.

   2. En la página Instances, haga clic en el nombre de la instancia de destino.

   3. En la esquina superior derecha de la página, haga clic en Connect to Instance.

   4. En la pestaña Metadata Management, haga doble clic en el nombre de la tabla users.

   5. Haga clic en Query table, introduzca la instrucción a continuación y luego haga clic en Run.

      SELECT * FROM test_schema.users;

   6. Consulte los resultados de la consulta.

      La consulta devuelve el resultado a continuación.

      El resultado muestra que el registro con el identificador 100001 se escribió correctamente en Hologres. Además, se agregaron dos columnas nuevas a la tabla de Hologres: house-points.house y house-points.points.

      Nota

      El mensaje enviado a ApsaraMQ for Kafka contiene solo una columna anidada: house-points. Sin embargo, dado que se especificó json.infer-schema.flatten-nested-columns.enable en la cláusula WITH, Realtime Compute for Apache Flink aplana automáticamente esta columna y utiliza las rutas de acceso de los campos anidados como nombres de las columnas nuevas.

Referencias

• Para obtener información sobre cómo usar Message Queue for Apache Kafka como tabla de origen o de resultados, consulte Message Queue for Apache Kafka.

• Si desea mejorar el rendimiento del trabajo mediante el ajuste del paralelismo y los recursos de los nodos, consulte Configure job deployment.