Realtime Compute for Apache Flink:Modifier une table source

Aperçu de la compatibilité

Détermination de la compatibilité

La méthode de vérification varie selon ce que vous modifiez :

• Modifications du schéma (DDL) : Le système évalue la compatibilité en se basant sur l'ensemble de la requête qui référence la table source, et non uniquement sur la définition de la table.

• Modifications de la configuration du connecteur (clause WITH) : Le système vérifie uniquement le paramètre du connecteur qui a été modifié, et non l'ensemble de la requête.

Modifications entièrement compatibles

Ces changements n'ont aucun impact sur les données d'état. Redémarrez le job à partir de l'état sans aucune action supplémentaire.

• Ajouter un champ qui n'est pas référencé dans la requête. Les opérateurs en aval ignorent le nouveau champ ; l'état reste inchangé.

• Modifier l'intervalle de watermark. La configuration du watermark n'affecte pas l'état des opérateurs.

Exemple — ajout d'un champ inutilisé :

-- Original
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Adding field d is fully compatible: d is not referenced in the SELECT statement.
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR,
  d INT
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

Modifications partiellement compatibles

Ces changements peuvent être appliqués en toute sécurité, mais une partie de l'état existant est perdue. Redémarrez le job à partir de l'état et attendez-vous à ce que les opérateurs concernés soient réinitialisés.

• Modifier un champ utilisé par certains opérateurs en aval, mais pas par d'autres. Les opérateurs qui dépendent du champ modifié perdent leur état accumulé. Ceux qui n'en dépendent pas ne sont pas affectés.

Exemple — renommage d'un champ avec une colonne calculée :

-- Original
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Renaming b to d and redefining b as a computed column is partially compatible.
-- MAX(c) is unaffected and retains its state.
-- The original SUM(b) is treated as deleted: its accumulated state is discarded.
-- The new SUM(b) is treated as added: it starts accumulating from zero when the job restarts.
CREATE TABLE MyTable (
  a INT,
  d BIGINT,
  c VARCHAR,
  b AS d + 1
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

Modifications incompatibles

Ces changements sont incompatibles avec les données d'état existantes. Ne redémarrez pas le job à partir de l'état.

Exemple — changement du type de connecteur :

-- Original
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Incompatible: changing the connector type from datagen to Kafka.
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'kafka',
  ...
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

Exemple — renommage de la table source :

-- Original
CREATE TABLE MyTable (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Incompatible: renaming the table from MyTable to MyTable2.
CREATE TABLE MyTable2 (
  a INT,
  b BIGINT,
  c VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT a, SUM(b), MAX(c) FROM MyTable2 GROUP BY a;

Exemple — ajout d'un champ affectant une jointure :

-- Original
CREATE TABLE MyTable1 (
  a INT,
  b BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE MyTable2 (
  c INT,
  d BIGINT
) WITH (
  'connector' = 'datagen'
);

SELECT * FROM MyTable1 JOIN MyTable2 ON c = d;

-- Incompatible: adding field e to MyTable2 changes the input schema of the join operator.
CREATE TABLE MyTable1 (
  a INT,
  b BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE MyTable2 (
  c INT,
  d BIGINT,
  e VARCHAR
) WITH (
  'connector' = 'datagen'
);

SELECT * FROM MyTable1 JOIN MyTable2 ON c = d;

Limitations

Certaines modifications qui semblent compatibles peuvent tout de même affecter l'état dans des conditions spécifiques :

• Modifications des clés primaires : La modification des clés primaires peut affecter la compatibilité de l'état des opérateurs en aval. Par exemple, changer la clé upsert peut provoquer des problèmes de compatibilité.

• Modifications des colonnes de sharding dans la clause WITH : La modification des clés primaires ou des colonnes de sharding dans la clause WITH peut affecter l'état de la table source. Par exemple, changer scan.incremental.snapshot.chunk.key-column pour le connecteur MySQL peut rendre les données d'état de la phase d'analyse complète indisponibles.

• Modifications du mode changelog : La modification du mode changelog dans la clause WITH peut affecter l'état des opérateurs en aval si les types d'événements envoyés en aval changent (par exemple, DELETE, UPDATE_BEFORE ou UPDATE_AFTER). Scénarios courants :

  • Changement du paramètre cdcMode pour le connecteur Hologres (contrôle si la capture des données modifiées (CDC) lit les journaux binaires).

  • Changement de la lecture des journaux binaires en mode UPSERT pour le connecteur MongoDB ou PostgreSQL.