Realtime Compute for Apache Flink:Crear y usar tablas materializadas

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>

Limitaciones

• Solo puede modificar la consulta de tablas materializadas creadas en VVR 11.1 o posterior.

• Al modificar una consulta, solo puede agregar columnas y modificar la lógica de cálculo. No puede cambiar el orden de las columnas existentes ni modificar sus definiciones.

  Operación	Compatible	Descripción
  Agregar una nueva columna	Sí	Puede añadir nuevas columnas al esquema manteniendo el orden de las columnas existentes.
  Modificar la lógica de cálculo de una columna existente (sin cambiar su nombre o tipo)	Sí	Puede modificar la lógica de cálculo, pero el nombre de la columna y el tipo de datos deben permanecer iguales.
  Cambiar el orden de las columnas existentes	No	El orden de las columnas es fijo. Para cambiarlo, debe eliminar y volver a crear la tabla materializada.
  Modificar el nombre o el tipo de datos de una columna existente	No	Debe eliminar y volver a crear la tabla materializada.

Ejemplo

1. Haga clic en Edit Table y modifique la consulta. El siguiente código proporciona un ejemplo:

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

2. Haga clic en Preview para ver una comparación del antes y el después.

   El cuadro de diálogo Materialized Table Change Details muestra los siguientes cambios: en el área SQL Statement, la instrucción SQL cambia de la original SELECT * a SELECT *, orders.price * orders.quantity AS total_price, e incluye una nueva condición WHERE orders.price * orders.quantity > 1000. En el área Table Columns, el nuevo campo total_price (DOUBLE) se resalta en verde.

3. Haga clic en OK. Puede ver la columna recién agregada y la lógica de consulta modificada en la pestaña Table Schema.

Actualizaciones incrementales

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;

Relleno de datos

Anteriormente, corregir los resultados del procesamiento de flujos para datos históricos requería desarrollar un trabajo por lotes independiente. Al usar tablas materializadas, puede rellenar directamente particiones de datos históricos. Este enfoque unifica el procesamiento por lotes y de streaming, lo que reduce los costos de desarrollo y operación y mantenimiento.

1. Haga clic en la tabla materializada deseada dentro de su catálogo.

2. En la pestaña Data Information, rellene los datos.

   Si definió una columna de particionamiento cuando created the materialized table, se trata de una tabla particionada. De lo contrario, es una tabla no particionada.

   Tabla particionada

   En la sección Partitions, haga clic en Trigger Update si es la primera vez que rellena datos o si la partición requerida no existe. Si ya existen particiones, puede seleccionar una partición específica y hacer clic en Refresh en la columna Actions.

   Haga clic en la pestaña Data Information y realice las operaciones en el área Data Partitions en la parte inferior de la página.

   Parámetros

   • Columna de particionamiento: La columna de particionamiento de la tabla. Por ejemplo, si introduce 20241201, se rellenarán todos los datos de la partición ds=20241201.

   • Task Name: El nombre de la tarea de relleno de datos.

   • Update Range (Opcional): Especifica si se deben aplicar actualizaciones en cascada a las tablas materializadas descendentes. A partir de la tabla actual, se actualizan todas las tablas materializadas en el linaje de datos. La profundidad máxima admitida del linaje descendente es de 6 niveles.

     Nota

     • Al actualizar una tabla particionada, las tablas materializadas descendentes deben tener exactamente las mismas columnas de particionamiento que la tabla inicial. De lo contrario, la operación de actualización fallará.

     • Si una actualización falla para alguna tabla materializada en el linaje, todos los nodos descendentes subsiguientes también fallarán.

   • Deployment Target: Puede seleccionar una cola o un clúster de sesiones. La selección predeterminada es default-queue.

   Tabla no particionada

   En la sección Data Detail, haga clic en Refresh.

   Parámetros

   • Task Name: El nombre de la tarea de relleno de datos.

   • Update Range: Esta opción no está disponible para tablas no particionadas.

     Nota

     • Durante una actualización, los datos descendentes se actualizan completamente.

     • Si una actualización falla para alguna tabla materializada en el linaje, todos los nodos descendentes subsiguientes también fallarán.

     • Las actualizaciones en cascada no son compatibles si la tabla inicial es una tabla no particionada actualizada por un trabajo de streaming.

   • Deployment Target: Puede seleccionar una cola o un clúster de sesiones. La selección predeterminada es default-queue.

3. Relleno programado y por lotes.

   Puede usar Task Orchestration para crear un flujo de trabajo para la tabla materializada y ejecutar trabajos de relleno según una programación. También puede usar la función de relleno de datos del flujo de trabajo para rellenar datos en bloque durante un intervalo de tiempo específico.

Modificar la actualización de los datos

1. En el catálogo correspondiente, haga clic en la base de datos de la materialized table y luego haga clic en la materialized table deseada.

2. En la esquina superior derecha, haga clic en Modify Freshness.

   • Si la tabla materializada no tiene clave principal, no puede cambiar su método de actualización entre streaming y lotes. El sistema utiliza un trabajo de streaming para valores de actualización de datos menores a 30 minutos y un trabajo por lotes para valores de 30 minutos o más. Por lo tanto, no se permite cambiar la actualización cruzando este umbral de 30 minutos para tablas sin clave principal.

   • Si la tabla ascendente es una tabla materializada, asegúrese de que la actualización de los datos de la tabla descendente sea un múltiplo entero positivo de la actualización de los datos de la tabla ascendente.

   • La actualización máxima de los datos es de un día.

Consultar el linaje de datos

En el panel de navegación izquierdo, vaya a O&M > Data Lineage para ir a la página de linaje de datos de tablas materializadas. En esta página, puede ver las relaciones de linaje entre todas las tablas materializadas. También puede realizar operaciones como Start/Stop Update y Modify Freshness directamente en una tabla materializada. Haga clic en Details para navegar a la página de detalles de la tabla materializada correspondiente.

Haga clic en un nodo de tabla materializada para expandir su panel de detalles. El panel muestra el valor de Data Freshness, Latest Update Time y Materialized Table Status, y proporciona una acción Trigger Update.