Realtime Compute for Apache Flink:Création et utilisation de tables matérialisées

CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-- Primary key constraint
[([CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED)]
[COMMENT table_comment]
-- Partition key
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
-- With options
[WITH (key1=val1, key2=val2, ...)]
-- Data freshness
FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
-- Refresh mode
[REFRESH_MODE = { CONTINUOUS | FULL }]
AS  <select_statement>

Limites

• Vous pouvez modifier la requête uniquement pour les tables matérialisées créées dans VVR 11.1 ou version ultérieure.

• Lors de la modification d'une requête, vous pouvez uniquement ajouter des colonnes et modifier la logique de calcul. Vous ne pouvez pas modifier l'ordre des colonnes existantes ni leurs définitions.

  Opération	Prise en charge	Description
  Ajouter une nouvelle colonne	Oui	Vous pouvez ajouter de nouvelles colonnes au schéma tout en conservant l'ordre des colonnes existantes.
  Modifier la logique de calcul d'une colonne existante (sans changer son nom ni son type)	Oui	Vous pouvez modifier la logique de calcul, mais le nom de la colonne et le type de données doivent rester identiques.
  Modifier l'ordre des colonnes existantes	Non	L'ordre des colonnes est fixe. Pour le modifier, vous devez supprimer et recréer la table matérialisée.
  Modifier le nom ou le type de données d'une colonne existante	Non	Vous devez supprimer et recréer la table matérialisée.

Exemple

1. Cliquez sur Edit Table et modifiez la requête. Le code suivant fournit un exemple :

   ALTER MATERIALIZED TABLE `paimon`.`default`.`mt-orders`
       AS
       SELECT
         *,
         price * quantity AS total_price
       FROM orders
       WHERE price * quantity > 1000
   ;

2. Cliquez sur Preview pour afficher une comparaison avant/après.

   La boîte de dialogue Materialized Table Change Details affiche les modifications suivantes : dans la zone SQL Statement, l'instruction SQL passe de la commande originale SELECT * à SELECT *, orders.price * orders.quantity AS total_price, et inclut une nouvelle condition WHERE orders.price * orders.quantity > 1000. Dans la zone Table Columns, le nouveau champ total_price (DOUBLE) est mis en surbrillance en vert.

3. Cliquez sur OK. Vous pouvez afficher la colonne nouvellement ajoutée et la logique de requête modifiée sous l'onglet Table Schema.

Mises à jour incrémentielles

CREATE MATERIALIZED TABLE mt_shipped_orders (
    PRIMARY KEY (order_id) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT 
    order_id,
    COALESCE(customer_id, 'Unknown') AS customer_id,
    CAST(order_amount AS DECIMAL(10, 2)) AS order_amount,
    CASE 
        WHEN status = 'shipped' THEN 'Completed'
        WHEN status = 'pending' THEN 'In Progress'
        ELSE 'Unknown'
    END AS order_status,
    DATE_FORMAT(order_ts, 'yyyyMMdd') AS order_date,
    UDSF_ProcessFunction(notes) AS notes
FROM 
    orders
WHERE
    status = 'shipped';

CREATE MATERIALIZED TABLE mt_enriched_orders (
    PRIMARY KEY (order_id, order_tag) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
WITH o AS (
    SELECT
        order_id,
        product_id,
        quantity,
        proc_time,
        e.tag AS order_tag
    FROM 
        orders,
        LATERAL TABLE(UDTF_StringSplitFunction(tags, ',')) AS e(tag))
SELECT 
    o.order_id,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    p.price,
    o.quantity * p.price AS total_amount,
    order_tag
FROM o 
LEFT JOIN 
    product_info FOR SYSTEM_TIME AS OF PROCTIME() AS p
ON 
    o.product_id = p.product_id;

Rétrochargement des données

Auparavant, la correction des résultats du traitement de flux pour les données historiques nécessitait le développement d'un job batch distinct. En utilisant des tables matérialisées, vous pouvez effectuer directement un rétrochargement des partitions de données historiques. Cette approche unifie le traitement batch et streaming, ce qui réduit les coûts de développement et d'exploitation.

1. Cliquez sur la table matérialisée cible sous son catalogue.

2. Sous l'onglet Data Information, effectuez le rétrochargement des données.

   Si vous avez défini une colonne de partitionnement lorsque vous avez créé la table matérialisée, il s'agit d'une table partitionnée. Sinon, il s'agit d'une table non partitionnée.

   Table partitionnée

   Dans la section Partitions, cliquez sur Trigger Update s'il s'agit de la première fois que vous effectuez un rétrochargement des données ou si la partition requise n'existe pas. Si les partitions existent déjà, vous pouvez sélectionner une partition spécifique et cliquer sur Refresh dans la colonne Actions.

   Cliquez sur l'onglet Data Information et effectuez les opérations dans la zone Data Partitions en bas de la page.

   Paramètres

   • Colonne de partitionnement : la colonne de partitionnement de la table. Par exemple, si vous saisissez 20241201, toutes les données de la partition ds=20241201 font l'objet d'un rétrochargement.

   • Nom de la tâche : le nom de la tâche de rétrochargement des données.

   • Plage de mise à jour (facultatif) : indique s'il faut propager les mises à jour aux tables matérialisées en aval. À partir de la table actuelle, toutes les tables matérialisées de la lignée des données sont mises à jour. La profondeur maximale prise en charge de la lignée en aval est de 6 niveaux.

     Remarque

     • Lors de la mise à jour d'une table partitionnée, les tables matérialisées en aval doivent avoir exactement les mêmes colonnes de partitionnement que la table de départ. Sinon, l'opération de mise à jour échoue.

     • Si une mise à jour échoue pour une table matérialisée quelconque de la lignée, tous les nœuds en aval suivants échouent également.

   • Cible de déploiement : vous pouvez sélectionner une file d'attente ou un cluster de session. La sélection par défaut est default-queue.

   Table non partitionnée

   Dans la section Data Detail, cliquez sur Refresh.

   Paramètres

   • Nom de la tâche : le nom de la tâche de rétrochargement des données.

   • Plage de mise à jour : cette option n'est pas disponible pour les tables non partitionnées.

     Remarque

     • Lors d'une mise à jour, les données en aval sont entièrement actualisées.

     • Si une mise à jour échoue pour une table matérialisée quelconque de la lignée, tous les nœuds en aval suivants échouent également.

     • Les mises à jour en cascade ne sont pas prises en charge si la table de départ est une table non partitionnée mise à jour par un job de streaming.

   • Cible de déploiement : vous pouvez sélectionner une file d'attente ou un cluster de session. La sélection par défaut est default-queue.

3. Planification et rétrochargement par lots.

   Vous pouvez utiliser Task Orchestration pour créer un workflow pour la table matérialisée afin d'exécuter des jobs de rétrochargement selon une planification. Vous pouvez également utiliser la fonctionnalité de rétrochargement des données du workflow pour effectuer un rétrochargement des données en masse pour une plage horaire spécifiée.

Modification de la fraîcheur des données

1. Sous le catalogue correspondant, cliquez sur la base de données materialized table, puis sur la materialized table cible.

2. Dans le coin supérieur droit, cliquez sur Modify Freshness.

   • Si la table matérialisée ne possède pas de clé primaire, vous ne pouvez pas basculer sa méthode de mise à jour entre streaming et lot. Le système utilise un job de streaming pour les valeurs de fraîcheur des données inférieures à 30 minutes et un job batch pour les valeurs de 30 minutes ou plus. Par conséquent, le franchissement de ce seuil de 30 minutes n'est pas autorisé pour les tables sans clé primaire.

   • Si la table en amont est une table matérialisée, assurez-vous que la fraîcheur des données de la table en aval est un multiple entier positif de la fraîcheur des données de la table en amont.

   • La fraîcheur maximale des données est d'un jour.

Consultation de la lignée des données

Dans le volet de navigation de gauche, accédez à O&M > Data Lineage pour accéder à la page de lignée des données pour les tables matérialisées. Sur cette page, vous pouvez afficher les relations de lignée entre toutes les tables matérialisées. Vous pouvez également effectuer des opérations telles que Start/Stop Update et Modify Freshness directement sur une table matérialisée. Cliquez sur Details pour accéder à la page de détails de la table matérialisée correspondante.

Cliquez sur un nœud de table matérialisée pour développer son panneau de détails. Le panneau affiche la valeur Data Freshness, l'heure de la Latest Update Time et le Materialized Table Status, et propose une action Trigger Update.