Realtime Compute for Apache Flink:Create and use materialized tables

Limits

• Only new materialized tables powered by VVR 11.1 or later are supported.

• Only adding columns and modifying computation logic are supported operations. See the table below for details:

  Operation	Supported	Description
  Add a column	Supported	Appends a new column to the materialized table while maintaining the column order.
  Modify computation logic	Supported	Modifies the computation logic of existing columns, without changing column names and types.
  Change the column order	Not supported	To change the column order, drop the materialized table and create a new one with the desired column order.
  Modify column names or types	Not supported	To modify the column names or types, drop the materialized table and create a new one with the desired column names or types.

Procedure

1. Click Edit Table and modify the query. Sample code:

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

2. Click Preview to check the modifications.

   

3. Click OK. Select the Table Schema tab to check the added columns and modified query.

Incrementally update a materialized table

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;

Backfill historical data

Traditionally, you need a separate batch job to correct stream processing results with all of the data from the previous day. With the materialized table, you can intuitively backfill data using the current job. This capability of materialized tables unifies batch and streaming processing, enhancing development and O&M efficiency.

1. In the Catalogs pane, click the target materialized table under your Apache Paimon catalog.

2. On the Data Information tab, refresh the data.

   If you have defined the partition key when creating the materialized table, the table is a partitioned table.

   Partitioned table

   In the upper-right corner of the Partitions section on the Table Information tab, click Trigger Update the first time you perform data backfilling or if no partitions are available. If a partition exists, find the corresponding partition and click Refresh in the Actions column.

   

   

   Parameters:

   • ds: the partition field of the table. For example, if you enter 20241201 in the ds field, all data with ds=20241201 is backfilled.

   • Task Name: the name of the data backfilling task.

   • Update Range (optional): specifies whether to cascade updates to the downstream materialized tables. You can use the partitioned table as the starting point to update all materialized tables. The maximum number of downstream layers is 6.

     Note

     • If you update a partitioned table, the partitioning columns of the downstream materialized table must be the same as the those of the partitioned table. Otherwise, the update operation fails.

     • If a materialized table fails to be updated, all downstream nodes fail.

   • Deployment Target: the object on which the table is deployed. You can select queues and session clusters. By default, default-queue is used.

   Non-partitioned table

   Select he Data Information tab, view the details of the materialized table, and click Refresh.

   

   Parameters:

   • Task Name: the name of the task for data backfilling.

   • Update Range: This parameter is not supported for non-partitioned tables.

     Note

     • All data in downstream tables will be refreshed.

     • If a materialized table fails to be updated, all downstream nodes fail.

     • If the system determines that the workload is streaming based on the data freshness of the base table and that the table is a non-partitioned table, cascade update is not supported.

   • Deployment Target: the object on which the table is deployed. You can select queues and session clusters. By default, default-queue is used.

Change data freshness

1. Click the materialized table database under the desired catalog and click the materialized table that you want to view.

2. In the upper-right corner of the details page of the materialized table, click Edit Freshness.

   • If the materialized table is a table that does not have a primary key, you cannot modify the streaming and batch properties of the task. For example, if you change the data freshness from 2 seconds to 1 hour, Realtime Compute for Apache Flink converts a streaming job into a batch one. If you change the data freshness from 1 hour to 2 seconds, Realtime Compute for Apache Flink converts a batch job into a streaming one. You cannot perform such operations. If the data freshness is less than 30 minutes, the job is classified as streaming. If the data freshness is equal to or greater than 30 minutes, the job is batch.

   • If the base table is a materialized table, make sure that the data freshness of the downstream table is 1 to N times that of the upstream table. N is a positive integer.

   • Data freshness cannot exceed 1 day.

View data lineage

On the Data Lineage page, view the data lineage among all materialized tables. You can start or stop the update of the materialized table and change its data freshness. You can also click Details in the upper-right corner of the page to go to the details page of the materialized table.