MaxCompute:Escritura de datos en MaxCompute con Flink

Contexto

• Modos de escritura compatibles

  El conector de Flink admite dos modos de escritura: upsert y insert. En el modo upsert, puede agrupar los flujos de datos de una de las siguientes maneras:

  • Agrupación por clave primaria

  • Agrupación por campo de partición

    La agrupación por campo de partición es adecuada cuando hay muchas particiones, pero puede provocar un desbalance de datos.

• Para conocer el procedimiento de escritura upsert y los parámetros recomendados, consulte Real-time data ingestion into a data warehouse.

• Especifique el modo de escritura mediante los parámetros del conector de Flink. Para consultar la lista completa de parámetros del conector, consulte Appendix: Flink connector parameters.

• Establezca el intervalo de checkpoint de los trabajos de escritura upsert de Flink en 3 minutos como mínimo. Los intervalos más cortos reducen la eficiencia de escritura y generan una gran cantidad de archivos pequeños.

• A continuación se muestra la correspondencia de tipos de datos entre Realtime Compute for Apache Flink y MaxCompute.

  Tipo de dato de Flink	Tipo de dato de MaxCompute
  CHAR(p)	CHAR(p)
  VARCHAR(p)	VARCHAR(p)
  STRING	STRING
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INT	INT
  BIGINT	BIGINT
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DECIMAL(p, s)	DECIMAL(p, s)
  DATE	DATE
  TIMESTAMP(9) WITHOUT TIME ZONE, TIMESTAMP_LTZ(9)	TIMESTAMP
  TIMESTAMP(3) WITHOUT TIME ZONE, TIMESTAMP_LTZ(3)	DATETIME
  BYTES	BINARY
  ARRAY<T>	ARRAY<T>
  MAP<K, V>	MAP<K, V>
  ROW	STRUCT

  Nota

  El tipo de dato TIMESTAMP de Flink no incluye información de zona horaria, mientras que el de MaxCompute sí. Esta diferencia puede generar una discrepancia de 8 horas. Para alinear las marcas de tiempo, utilice TIMESTAMP_LTZ(9).

  -- Flink SQL
  CREATE TEMPORARY TABLE odps_source(
    id BIGINT NOT NULL COMMENT 'ID',
    created_time TIMESTAMP NOT NULL COMMENT 'Creation time',
    updated_time TIMESTAMP_LTZ(9) NOT NULL COMMENT 'Update time',
  PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
  'connector' = 'maxcompute',
  ...
  );

Escritura de datos desde un clúster de Flink autogestionado

1. Requisitos previos: cree una tabla de MaxCompute.

   Primero debe crear una tabla de MaxCompute donde Flink escribirá los datos. El siguiente ejemplo ilustra este proceso mediante la creación de dos tablas: una tabla delta sin particiones y una tabla con particiones. Para obtener información sobre la configuración de propiedades de tabla, consulte delta table parameters.

   -- Create a non-partitioned delta table.
   CREATE TABLE mf_flink_tt (
     id BIGINT not null,
     name STRING,
     age INT,
     status BOOLEAN, primary key (id)
   )
   tblproperties ("transactional"="true", 
                  "write.bucket.num" = "64", 
                  "acid.data.retain.hours"="12") ;
   
   --Create a partitioned delta table.
   CREATE TABLE mf_flink_tt_part (
     id BIGINT not null,
     name STRING,
     age INT,
     status BOOLEAN, 
     primary key (id)
   )
     partitioned by (dd string, hh string) 
     tblproperties ("transactional"="true", 
                    "write.bucket.num" = "64", 
                    "acid.data.retain.hours"="12") ;
   

2. Configure un clúster de Flink de código abierto. El conector es compatible con Flink 1.13, 1.15, 1.16 y 1.17. Seleccione el conector de Flink que corresponda a su versión de Flink:

   • Conector de Flink para Flink 1.13

   • Conector de Flink para Flink 1.15

   • Conector de Flink para Flink 1.16 y 1.17

   Nota

   • El conector de Flink para Flink 1.16 es compatible con Flink 1.17.

   • Este tema utiliza como ejemplo el conector de Flink para Flink 1.13. Descargue y descomprima el paquete.

3. Descargue el conector de Flink y agréguelo al paquete de su clúster de Flink.

   1. Descargue el paquete JAR del conector de Flink en su entorno local.

   2. Agregue el paquete JAR del conector de Flink al directorio lib del paquete de instalación de Flink descomprimido.

      mv flink-connector-odps-1.13-shaded.jar $FLINK_HOME/lib/flink-connector-odps-1.13-shaded.jar

4. Inicie el servicio de la instancia de Flink.

   cd $FLINK_HOME/bin
   ./start-cluster.sh

5. Inicie el cliente de Flink.

   cd $FLINK_HOME/bin
   ./sql-client.sh

6. Cree una tabla de Flink y configure los parámetros del conector de Flink.

   Puede crear una tabla de Flink y configurar sus parámetros mediante Flink SQL o la API DataStream. A continuación se presentan ejemplos básicos de ambos enfoques.

   Flink SQL

   1. En el editor de Flink SQL, ejecute los siguientes comandos para crear una tabla y configurar los parámetros.

      -- Register a non-partitioned table in Flink SQL.
      CREATE TABLE mf_flink (
        id BIGINT,
        name STRING,
        age INT,
        status BOOLEAN,
        PRIMARY KEY(id) NOT ENFORCED
      ) WITH (
        'connector' = 'maxcompute',
        'table.name' = 'mf_flink_tt',
        'sink.operation' = 'upsert',
        'odps.access.id'='LTAI****************',
        'odps.access.key'='********************',
        'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
        'odps.project.name'='mf_mc_bj'
      );
      
      -- Register a partitioned table in Flink SQL.
      CREATE TABLE mf_flink_part (
        id BIGINT,
        name STRING,
        age INT,
        status BOOLEAN,
        dd STRING,
        hh STRING,
        PRIMARY KEY(id) NOT ENFORCED
      ) PARTITIONED BY (`dd`,`hh`)
      WITH (
        'connector' = 'maxcompute',
        'table.name' = 'mf_flink_tt_part',
        'sink.operation' = 'upsert',
        'odps.access.id'='LTAI****************',
        'odps.access.key'='********************',
        'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
        'odps.project.name'='mf_mc_bj'
      );

   2. Escriba datos en la tabla de Flink y consulte la tabla de MaxCompute para comprobar el resultado.

      -- Insert data into the non-partitioned table in the Flink SQL client.
      INSERT INTO mf_flink VALUES (1,'Danny',27, false);
      
      -- Query result in MaxCompute.
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 27   | false  |
      +------------+------+------+--------+
      
      -- Insert data into the non-partitioned table in the Flink SQL client to update the record.
      INSERT INTO mf_flink VALUES (1,'Danny',28, false);
      -- Query result in MaxCompute.
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 28   | false  |
      +------------+------+------+--------+
      
      -- Insert data into the partitioned table in the Flink SQL client.
      INSERT INTO mf_flink_part VALUES (1,'Danny',27, false, '01','01');
      -- Query result in MaxCompute.
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 27   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+
      
      -- Insert data into the partitioned table in the Flink SQL client to update the record.
      INSERT INTO mf_flink_part VALUES (1,'Danny',30, false, '01','01');
      -- Query result in MaxCompute.
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 30   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+

   DataStream API

   1. Para utilizar la API DataStream, agregue la siguiente dependencia.

      <dependency>
        <groupId>com.aliyun.odps</groupId>
        <artifactId>flink-connector-maxcompute</artifactId>
                  <version>xxx</version>
                  <scope>system</scope>
                  <systemPath>${mvn_project.basedir}/lib/flink-connector-maxcompute-xxx-shaded.jar</systemPath>
      </dependency>

      Nota

      Reemplace "xxx" por el número de versión real.

   2. El siguiente código de ejemplo muestra cómo crear una tabla y configurar los parámetros.

      package com.aliyun.odps.flink.examples;
      
      import org.apache.flink.configuration.Configuration;
      import org.apache.flink.odps.table.OdpsOptions;
      import org.apache.flink.odps.util.OdpsConf;
      import org.apache.flink.odps.util.OdpsPipeline;
      import org.apache.flink.streaming.api.datastream.DataStream;
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.table.api.Table;
      import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
      import org.apache.flink.table.data.RowData;
      
      public class Examples {
      
          public static void main(String[] args) throws Exception {
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              env.enableCheckpointing(120 * 1000);
      
              StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);
      
              Table source = streamTableEnvironment.sqlQuery("SELECT * FROM source_table");
              DataStream<RowData> input = streamTableEnvironment.toAppendStream(source, RowData.class);
      
              Configuration config = new Configuration();
              config.set(OdpsOptions.SINK_OPERATION, "upsert");
              config.set(OdpsOptions.UPSERT_COMMIT_THREAD_NUM, 8);
              config.set(OdpsOptions.UPSERT_MAJOR_COMPACT_MIN_COMMITS, 100);
      
              OdpsConf odpsConfig = new OdpsConf("accessid",
                      "accesskey",
                      "endpoint",
                      "project",
                      "tunnel endpoint");
      
              OdpsPipeline.Builder builder = OdpsPipeline.builder();
              builder.projectName("sql2_isolation_2a")
                      .tableName("user_ledger_portfolio")
                      .partition("")
                      .configuration(config)
                      .odpsConf(odpsConfig)
                      .sink(input, false);
              env.execute();
          }
      }

Escritura de datos desde Flink totalmente administrado

1. Requisitos previos: cree una tabla de MaxCompute.

   Debe crear una tabla de MaxCompute de destino para los datos de Flink. El siguiente ejemplo muestra cómo crear una tabla delta.

   SET odps.sql.type.system.odps2=true;
   DROP TABLE mf_flink_upsert;
   CREATE TABLE mf_flink_upsert (
     c1 int not null, 
     c2 string, 
     gt timestamp,
     primary key (c1)
   ) 
     PARTITIONED BY (ds string)
     tblproperties ("transactional"="true",
                    "write.bucket.num" = "64", 
                    "acid.data.retain.hours"="12") ;

2. El conector de Flink viene precargado en Flink totalmente administrado, por lo que no requiere instalación manual. Puede consultar los detalles del conector en la consola de Realtime Compute for Apache Flink.

3. Cree una tabla de Flink, genere datos de Flink en tiempo real mediante un trabajo de Flink SQL y, tras finalizar el desarrollo, despliegue el trabajo.

   En la página de desarrollo de trabajos de Flink, cree y edite un trabajo SQL. El siguiente ejemplo define una tabla de origen que genera datos aleatorios, una tabla de resultados conectada a MaxCompute y una instrucción INSERT para transferir los datos. Para obtener más información sobre cómo desarrollar un trabajo SQL, consulte Job development map.

   -- Create a Flink source table.
   CREATE TEMPORARY TABLE fake_src_table
   (
       c1 int,
       c2 VARCHAR,
       gt AS CURRENT_TIMESTAMP
   ) WITH (
     'connector' = 'faker',
     'fields.c2.expression' = '#{superhero.name}',
     'rows-per-second' = '100',
     'fields.c1.expression' = '#{number.numberBetween ''0'',''1000''}'
   );
   
   -- Create a temporary result table in Flink.
   CREATE TEMPORARY TABLE test_c_d_g 
   (
       c1 int,
       c2 VARCHAR,
       gt TIMESTAMP,
       ds varchar,
       PRIMARY KEY(c1) NOT ENFORCED
    ) PARTITIONED BY(ds)
    WITH (
       		'connector' = 'maxcompute',
       		'table.name' = 'mf_flink_upsert',
       		'sink.operation' = 'upsert',
       		'odps.access.id'='LTAI****************',
       		'odps.access.key'='********************',
       		'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
       		'odps.project.name'='mf_mc_bj',
       		'upsert.write.bucket.num'='64'
   );
   
   -- Flink computation logic.
   INSERT INTO test_c_d_g
   SELECT  c1 AS c1,
           c2 AS c2,
           gt AS gt,
           date_format(gt, 'yyyyMMddHH') AS ds
   FROM    fake_src_table;

   Parámetros:

   odps.end.point: utilice el endpoint de red interna de la región correspondiente.

   upsert.write.bucket.num: este valor debe coincidir con el de la propiedad write.bucket.num de la tabla delta creada en MaxCompute.

4. Consulte la tabla de MaxCompute para comprobar que los datos se escribieron correctamente.

   SELECT * FROM mf_flink_upsert WHERE ds=2023061517;
   
   -- Your results may differ because the source data is randomly generated.
   +------+----+------+----+
   | c1   | c2 | gt   | ds |
   +------+----+------+----+
   | 0    | Skaar | 2023-06-16 01:59:41.116 | 2023061517 |
   | 21   | Supah Century | 2023-06-16 01:59:59.117 | 2023061517 |
   | 104  | Dark Gorilla Grodd | 2023-06-16 01:59:57.117 | 2023061517 |
   | 126  | Leader | 2023-06-16 01:59:39.116 | 2023061517 |
   

Apéndice: parámetros del conector de Flink

• Parámetros básicos

  Parámetro	Obligatorio	Valor predeterminado	Descripción
  connector	Sí	—	Establezca el tipo de conector en MaxCompute.
  odps.project.name	Sí	—	Nombre del proyecto de MaxCompute.
  odps.access.id	Sí	—	AccessKey ID de su cuenta. Consulte la página AccessKey Pair.
  odps.access.key	Sí	—	AccessKey secret de su cuenta. Consulte la página AccessKey Pair.
  odps.end.point	Sí	—	Endpoint de MaxCompute. Para consultar la lista de endpoints por región, consulte Endpoints.
  odps.tunnel.end.point	No	—	Endpoint público del servicio Tunnel. De forma predeterminada, las solicitudes se enrutan automáticamente al endpoint de Tunnel adecuado. Configure este parámetro para utilizar un endpoint específico y deshabilitar el enrutamiento automático. Para obtener más información sobre los endpoints de Tunnel en distintas regiones y redes, consulte Endpoints.
  odps.tunnel.quota.name	No	—	Nombre de la cuota de Tunnel utilizada para acceder a MaxCompute.
  table.name	Sí	—	Nombre de la tabla de MaxCompute, con el formato [project.][schema.]table.
  odps.namespace.schema	No	false	Indica si se utiliza el modelo de tres capas. Para obtener más información sobre este modelo, consulte Schema operations.
  sink.operation	Sí	insert	Tipo de escritura. Los valores válidos son insert o upsert. Nota El modo upsert solo es compatible con las tablas delta de MaxCompute.
  sink.parallelism	No	—	Paralelismo de escritura. Si no se configura, toma de forma predeterminada el paralelismo de la fuente ascendente. Nota Asegúrese de que la propiedad de tabla write.bucket.num sea un múltiplo entero del valor configurado. Esto optimiza el rendimiento de escritura y maximiza el ahorro de memoria en el nodo Sink.
  sink.meta.cache.time	No	400	Tamaño de la caché de metadatos.
  sink.meta.cache.expire.time	No	1200	Tiempo de expiración de la caché de metadatos, en segundos.
  sink.coordinator.enable	No	true	Indica si se habilita el modo coordinador.

• Parámetros de partición

  Parámetro	Obligatorio	Valor predeterminado	Descripción
  sink.partition	No	—	Nombre de la partición en la que se escriben los datos. Si utiliza particionamiento dinámico, este parámetro indica el nombre de la partición principal de las particiones dinámicas.
  sink.partition.default-value	No	__DEFAULT_PARTITION__	Nombre de la partición predeterminada cuando se utiliza particionamiento dinámico.
  sink.dynamic-partition.limit	No	100	Número máximo de particiones en las que se puede escribir de forma simultánea durante un solo checkpoint con particionamiento dinámico. Nota Un aumento significativo de este valor puede provocar errores de falta de memoria (OOM) en el nodo sink. Si la cantidad de particiones simultáneas supera este límite, el trabajo fallará.
  sink.group-partition.enable	No	false	Indica si se agrupa por partición al utilizar particionamiento dinámico.
  sink.partition.assigner.class	No	—	Clase de implementación de PartitionAssigner.

• Parámetros de escritura en modo FileCached

  Utilice el modo de caché de archivos para trabajos con una gran cantidad de particiones dinámicas. Los siguientes parámetros permiten configurar este modo.

  Parámetro	Obligatorio	Valor predeterminado	Descripción
  sink.file-cached.enable	No	false	Habilita el modo FileCached. Se recomienda para trabajos con una gran cantidad de particiones dinámicas. false: el modo FileCached está deshabilitado. true: el modo FileCached está habilitado. Nota Cuando la cantidad de particiones dinámicas es grande, puede utilizar el modo de caché de archivos.
  sink.file-cached.tmp.dirs	No	./local	Directorio predeterminado de la caché de archivos en modo FileCached.
  sink.file-cached.writer.num	No	16	Número de hilos simultáneos de carga de datos para una sola tarea en modo FileCached. Nota No aumente significativamente el valor de este parámetro. Si se escriben demasiadas particiones de forma simultánea, es probable que ocurran errores OOM.
  sink.bucket.check-interval	No	60000	Intervalo para comprobar el tamaño del archivo en modo FileCached. Unidad: milisegundos (ms).
  sink.file-cached.rolling.max-size	No	16 M	Tamaño máximo de un solo archivo de caché. Cuando un archivo supera este tamaño, se carga automáticamente.
  sink.file-cached.memory	No	64 M	Tamaño máximo de memoria fuera del heap utilizado para la escritura de archivos en modo FileCached.
  sink.file-cached.memory.segment-size	No	128 KB	Tamaño del búfer utilizado para la escritura de archivos en modo FileCached.
  sink.file-cached.flush.always	No	true	Indica si se utiliza la caché para la escritura de archivos en modo FileCached.
  sink.file-cached.write.max-retries	No	3	Número de reintentos para cargar datos en modo FileCached.

• Parámetros de escritura Insert o Upsert

  Parámetros de escritura Upsert

  Parámetro	Obligatorio	Valor predeterminado	Descripción
  upsert.writer.max-retries	No	3	Número de reintentos después de que un escritor upsert no logra escribir datos en un bucket.
  upsert.writer.buffer-size	No	64 MB	Tamaño de la caché para un solo escritor upsert en Flink. Nota Cuando la suma de los tamaños de búfer de todos los buckets alcanza el umbral preestablecido, el sistema activa automáticamente una operación de flush para actualizar los datos en el servidor. Un escritor upsert escribe datos en varios buckets de forma simultánea. Se recomienda aumentar el valor de este parámetro para mejorar la eficiencia de escritura. Si se escriben datos en una gran cantidad de particiones, pueden ocurrir errores OOM. En ese caso, reduzca el valor de este parámetro.
  upsert.writer.bucket.buffer-size	No	1 MB	Tamaño de la caché para un solo bucket en Flink. Si los recursos de memoria del servidor de Flink son insuficientes, reduzca el valor de este parámetro.
  upsert.write.bucket.num	Sí	—	El número de buckets de la tabla de destino debe ser igual al valor de write.bucket.num.
  upsert.write.slot-num	No	1	Número de slots de Tunnel utilizados por una sola sesión.
  upsert.commit.max-retries	No	3	Número de reintentos para el commit de una sesión upsert.
  upsert.commit.thread-num	No	16	Paralelismo del commit de una sesión upsert. No establezca un valor demasiado alto. Una gran cantidad de commits simultáneos incrementa el consumo de recursos, lo que puede causar problemas de rendimiento o un uso excesivo de recursos.
  upsert.major-compact.min-commits	No	100	Número mínimo de commits necesarios para activar una compactación mayor.
  upsert.commit.timeout	No	600	Tiempo de espera para el commit de una sesión upsert. Unidad: segundos (s).
  upsert.major-compact.enable	No	false	Indica si se habilita la compactación mayor.
  upsert.flush.concurrent	No	2	Número máximo de buckets en los que se pueden escribir datos de forma simultánea dentro de una sola partición. Nota Cuando se vacían los datos de un bucket, se ocupa un slot de Tunnel.

  Nota

  Para obtener más información sobre las configuraciones de parámetros recomendadas para escrituras upsert, consulte Recommended parameter configurations for upsert writes.

  Parámetros de escritura Insert

  Parámetro	Obligatorio	Valor predeterminado	Descripción
  insert.commit.thread-num	No	16	Paralelismo de una sesión de commit.
  insert.arrow-writer.enable	No	false	Indica si se utiliza el formato Arrow.
  insert.arrow-writer.batch-size	No	512	Número máximo de filas en un lote de Arrow.
  insert.arrow-writer.flush-interval	No	100000	Intervalo de vaciado del escritor. Unidad: milisegundos (ms).
  insert.writer.buffer-size	No	64 MB	Tamaño de la caché del escritor con búfer.