Realtime Compute for Apache Flink:Gestionar catálogos Paimon

• Crear una tabla con la instrucción CREATE TABLE

  Ejecute el siguiente comando en el editor SQL desde Development > Scripts.

  El siguiente ejemplo crea una tabla particionada en la base de datos my_db del catálogo my-catalog.

  -- Replace my-catalog with the name of your Paimon catalog.
  -- Replace my_db with the name of the database that you want to use.
  -- You can also replace my_tbl with a custom name in English.
  CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
    dt STRING,
    shop_id BIGINT,
    user_id BIGINT,
    num_orders INT,
    total_amount INT,
    PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
  ) PARTITIONED BY (dt) WITH (
    'bucket' = '4'
  );

  Para obtener más información, consulte Apache Paimon connector y Primary key and append-only tables.

• Crear una tabla con Flink CDC data ingestion

  Los trabajos de ingesta de datos de Flink CDC sincronizan tanto los datos como los cambios de esquema. Puede usar la ingesta de datos de Flink CDC para replicar tablas desde orígenes como MySQL y Kafka hacia un catálogo Paimon.

  Dado que la ingesta de datos de Flink CDC replica datos, debe deploy e start el trabajo. Para obtener más información, consulte Job development overview y Start a job.

  Nota

  La ingesta de datos de Flink CDC admite configurar propiedades de tabla física en los parámetros. Cuando se crea la tabla de destino, se aplican las propiedades correspondientes a dicha tabla. Al iniciar el trabajo, estos parámetros se aplican a las tablas descendentes que se sincronizan. Para obtener más información sobre las propiedades de tabla compatibles, consulte Data ingestion for the Paimon connector.

  • Replicar una tabla específica

    Por ejemplo, el siguiente trabajo crea la tabla Paimon my_db a partir del esquema de la tabla mysql.tpcds.web_sales y replica los datos. Además, establece el número de buckets de la tabla Paimon en 4 y usa el productor de changelog de entrada.

    source:
      type: mysql
      using.built-in-catalog: mysql
      tables: tpcds.web_sales
      server-id: <server id>
      # (Optional) Synchronize data from tables created during the incremental phase.
      scan.binlog.newly-added-table.enabled: true
      # (Optional) Synchronize table comments and column comments.
      include-comments.enabled: true
      # (Optional) Assign unbounded chunks first to prevent possible TaskManager out-of-memory errors.
      scan.incremental.snapshot.unbounded-chunk-first.enabled: true
      # (Optional) Enable parse filtering to accelerate reads.
      scan.only.deserialize.captured.tables.changelog.enabled: true
    
    sink:
      type: paimon
      using.built-in-catalog: paimon-dlf-catalog
      table.properties.bucket: "4"
      table.properties.changelog-producer: input
      # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
      commit.user: your_job_name
      # (Optional) Enable deletion vectors to improve read performance.
      table.properties.deletion-vectors.enabled: true
    
    route:
      - source-table: tpcds.web_sales
        sink-table: my_db.web_sales

  • Replicar una base de datos completa

    Por ejemplo, el siguiente trabajo crea una tabla Paimon por cada tabla de la base de datos my-catalog con base en su esquema, escribe las tablas en my_db y replica los datos. También configura el productor de changelog de entrada para las tablas Paimon.

    source:
      type: mysql
      using.built-in-catalog: mysql-catalog
      # Match all tables in the tpcds database. Single quotes preserve the backslash.
      tables: 'tpcds.\.*'
      server-id: <server id>
    
    sink:
      type: paimon
      using.built-in-catalog: paimon-dlf-catalog
      table.properties.changelog-producer: input
    
    route:
      - source-table: 'tpcds.\.*'
        sink-table: "my_db.<>"
        replace-symbol: "<>"
    

  • Replicar cambios de tipo de columna

    Las tablas Paimon creadas por trabajos de ingesta de datos de Flink CDC admiten no solo columnas nuevas, sino también ciertos cambios de tipo de columna. Puede elegir si habilita la normalización de tipos según sus necesidades.

    • Replicación de tipos predeterminada

      Las columnas de una tabla Paimon creada por un trabajo de ingesta de datos de Flink CDC usan los mismos tipos que las columnas de la tabla de origen. Se admiten los siguientes cambios de tipo de columna:

      • Tipos enteros: TINYINT, SMALLINT, INT y BIGINT se pueden promover a un tipo entero de precisión igual o mayor. (TINYINT < SMALLINT < INT < BIGINT).

      • Tipos de punto flotante: FLOAT y DOUBLE se pueden promover a un tipo de punto flotante de precisión igual o mayor. (FLOAT < DOUBLE).

      • Tipos de cadena: CHAR, VARCHAR y STRING se pueden promover a un tipo de cadena de precisión igual o mayor.

    • Uso de la normalización de tipos en CTAS/CDAS

      Para permitir cambios de tipo más flexibles y evitar que los trabajos fallen por evoluciones de esquema compatibles, establezca 'enableTypeNormalization' = 'true' en la cláusula WITH de su instrucción CTAS/CDAS. En este modo, Flink normaliza los tipos de datos durante la evolución del esquema. Un trabajo solo fallará si los tipos antiguo y nuevo no se pueden normalizar al mismo tipo subyacente. Reglas:

      • TINYINT, SMALLINT, INT y BIGINT se normalizan a BIGINT.

      • FLOAT y DOUBLE se normalizan a DOUBLE.

      • CHAR, VARCHAR y STRING se normalizan a STRING.

      • Los demás tipos de datos no se normalizan.

    • Impacto de la normalización:

      Cuando la normalización de tipos está activa, los tipos de datos almacenados en la tabla Paimon serán el tipo normalizado. Por ejemplo, si las columnas de origen c1 (SMALLINT) y c2 (INT) se normalizan a BIGINT, tanto c1 como c2 se almacenarán como BIGINT en la tabla Paimon.

    • Ejemplos:

      • Éxito: si SMALLINT en el origen cambia a INT, ambos tipos se normalizan a BIGINT. Este es un cambio compatible y el trabajo continúa con normalidad.

      • Error: si FLOAT en el origen cambia a BIGINT, sus tipos normalizados (DOUBLE y BIGINT) son incompatibles, lo que provocará una excepción.

Ejecute los siguientes comandos en el editor SQL desde Development > Scripts.

ALTER TABLE my_table SET (
  'write-buffer-size' = '256 MB',
  'write-buffer-spillable' = 'true'
);

ALTER TABLE my_table RENAME TO my_table_new;

ALTER TABLE my_table RENAME c0 TO c1;

ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

ALTER TABLE my_table MODIFY col_a DOUBLE;

Ejecute el siguiente comando en el editor SQL desde Development > Scripts.

-- Replace my-catalog with the name of your Paimon catalog.
-- Replace my_db with the name of the database that you want to use.
-- Replace my_tbl with the name of the Paimon catalog table that you created.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

Si se devuelve el mensaje The following statement has been executed successfully!, la tabla Paimon se eliminó correctamente.