Todos los productos
Search

Este producto

    Realtime Compute for Apache Flink:Tablas de clave primaria y de solo anexión

    Centro de documentos

    Realtime Compute for Apache Flink:Tablas de clave primaria y de solo anexión

    Última actualización: Sep 03, 2026

    Apache Paimon admite dos tipos de tablas: tablas de clave primaria y tablas de solo anexión. Cada tipo tiene una semántica distinta para escrituras, actualizaciones y consumo mediante streaming. Elija el tipo que se ajuste a los requisitos de su pipeline de datos.

    Selección de un tipo de tabla

    Escenario

    Configuración recomendada

    Sincronización CDC, upserts estándar

    Tabla de clave primaria, motor deduplicate

    Deduplicación donde solo importa el primer registro

    Tabla de clave primaria, motor first-row

    Agregación en tiempo real (totales acumulados, seguimiento de máximos/mínimos)

    Tabla de clave primaria, motor aggregation

    Ensamblaje de tablas anchas desde varios orígenes de streaming

    Tabla de clave primaria, motor partial-update

    Ingesta de registros, streaming de eventos

    Tabla de solo anexión (escalable)

    Reemplazo de cola de mensajes con preservación de orden

    Tabla de cola de anexión

    Tablas de clave primaria

    Una tabla de clave primaria requiere una o más claves primarias, las cuales se especifican al crear la tabla. Los registros con las mismas claves primarias se fusionan según el motor de fusión configurado.

    CREATE TABLE T (
  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'
);

    En este ejemplo, la clave de partición es dt, las claves primarias son dt, shop_id y user_id, y la tabla utiliza 4 buckets fijos.

    Modos de bucket

    Un bucket es la unidad mínima de lectura y escritura en una tabla Paimon. Las particiones (o la tabla completa en el caso de tablas sin particionar) se subdividen en buckets, lo que permite lecturas y escrituras paralelas.

    Modo

    Configuración

    Notas

    Modo de bucket dinámico (predeterminado)

    Omita bucket o establezca 'bucket' = '-1'

    No admite escrituras simultáneas desde varios despliegues de Flink; admite actualizaciones entre particiones

    Modo de bucket fijo

    Establezca 'bucket' = '<num>' (entero > 0)

    <num> es la cantidad de buckets para toda la tabla (sin particionar) o por partición (particionada); admite cambiar el recuento de buckets

    Dimensionamiento de buckets: Se recomienda configurar el tamaño total de datos en cada bucket en 2 GB y no especificar un valor superior a 5 GB. Muy pocos buckets limitan el paralelismo de escritura; demasiados generan una cantidad excesiva de archivos pequeños.

    Modo de bucket dinámico: cómo se actualizan los datos

    Las actualizaciones entre particiones ocurren cuando las claves primarias no incluyen todas las claves de partición. Paimon no puede determinar la partición de destino solo con la clave primaria, por lo que utiliza RocksDB para mantener un mapeo de clave primaria a partición/bucket. En tablas grandes, esto puede reducir significativamente el rendimiento en comparación con el modo de bucket fijo, y la inicialización del despliegue tarda más mientras el mapeo se carga en RocksDB.

    El comportamiento de actualización depende del motor de fusión:

    • deduplicate: elimina el registro existente e inserta uno nuevo en la partición de destino.

    • aggregation o partial-update: actualiza el registro existente en su lugar dentro de la partición actual.

    • first-row: conserva el registro existente y descarta el registro entrante.

    Las actualizaciones dentro de la partición ocurren cuando las claves primarias incluyen todas las claves de partición. Paimon puede determinar la partición a partir de la clave primaria, pero no el bucket, por lo que mantiene un índice en memoria que mapea las claves primarias a los buckets. Cada 100 millones de entradas de mapeo consumen aproximadamente 1 GB de memoria heap, la cual se asigna únicamente a las particiones que se están escribiendo activamente.

    Modo de bucket dinámico: asignación de buckets

    Los nuevos registros se escriben primero en los buckets existentes. Si todos los buckets han alcanzado su capacidad, se crea automáticamente un nuevo bucket.

    Configure este comportamiento con los siguientes parámetros en la cláusula WITH:

    Parámetro

    Descripción

    Valor predeterminado

    dynamic-bucket.target-row-num

    Máximo de registros por bucket

    2000000

    dynamic-bucket.initial-buckets

    Recuento inicial de buckets

    Paralelismo del operador Writer

    Modo de bucket fijo: asignación de buckets

    De forma predeterminada, Paimon asigna registros a los buckets mediante un hash de los valores de la clave primaria. Para usar un conjunto diferente de columnas, establezca bucket-key en la cláusula WITH. Especifique varias columnas separadas por comas. Las columnas en bucket-key deben ser un subconjunto de las claves primarias.

    Por ejemplo, 'bucket-key' = 'c1,c2' enruta los registros según el hash de c1 y c2.

    En el modo de bucket fijo, incluya todas las claves de partición en las claves primarias para evitar actualizaciones entre particiones.

    Cambio del número de buckets en el modo de bucket fijo

    El recuento de buckets determina el paralelismo de lectura y escritura. Para modificarlo en una tabla existente:

    1. Pause todos los despliegues que lean o escriban en la tabla.

    2. Cree un script y ejecute la siguiente sentencia SQL para configurar el parámetro de bucket:

      ALTER TABLE `<catalog-name>`.`<database-name>`.`<table-name>` SET ('bucket' = '<bucket-num>');

    3. Reorganice los datos ejecutando un despliegue por lotes.

      • Tabla sin particionar: Cree un Blank Batch Draft, pegue el siguiente código SQL, luego haga clic en Deploy y Start: ``sql INSERT OVERWRITE <catalog-name>.<database-name>.<table-name> SELECT * FROM <catalog-name>.<database-name>.<table-name>; ``

      • Tabla particionada: Ejecute un proceso por lotes por cada partición que desee reorganizar:

        INSERT OVERWRITE `<catalog-name>`.`<database-name>`.`<table-name>`
PARTITION (<partition-spec>)
SELECT * FROM `<catalog-name>`.`<database-name>`.`<table-name>`
WHERE <partition-condition>;

        Ejemplo — reorganizar la partición donde dt = '20240312' y hh = '08':

        INSERT OVERWRITE `<catalog-name>`.`<database-name>`.`<table-name>`
PARTITION (dt = '20240312', hh = '08')
SELECT * FROM `<catalog-name>`.`<database-name>`.`<table-name>`
WHERE dt = '20240312' AND hh = '08';

    4. Una vez que el despliegue por lotes finalice correctamente, reanude los despliegues pausados.

    Motores de fusión

    Cuando varios registros comparten las mismas claves primarias, Paimon los fusiona según la configuración de merge-engine.

    Si su flujo de entrada contiene registros desordenados, configure el parámetro sequence.field para controlar el orden de fusión. Consulte Manejo de datos desordenados .

    deduplicate (predeterminado)

    Paimon conserva solo el registro más reciente para cada conjunto de claves primarias y descarta los anteriores. Si el último registro es una operación DELETE, se eliminan todos los registros con esas claves primarias.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'deduplicate' -- Optional: deduplicate is the default
);

    Resultados de ejemplo:

    • Escriba +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), +I(1, 8.0, 'cherry') — la consulta devuelve (1, 8.0, 'cherry').

    • Escriba +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), -D(1, 4.0, 'banana') — la consulta no devuelve filas.

    Uso recomendado: Sincronización CDC, pipelines de upsert estándar.

    first-row

    Paimon conserva solo el primer registro para cada conjunto de claves primarias. Dado que los consumidores posteriores solo reciben eventos de changelog de tipo INSERT, la generación del changelog es más eficiente que con deduplicate.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'first-row'
);

    Resultado de ejemplo:

    Escriba +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), +I(1, 8.0, 'cherry') — la consulta devuelve (1, 2.0, 'apple').

    Notas de uso:

    • Establezca changelog-producer en lookup para habilitar el consumo mediante streaming.

    • El motor first-row no procesa cambios de tipo DELETE ni UPDATE_BEFORE. Para ignorarlos en lugar de generar un error, establezca 'first-row.ignore-delete' = 'true'.

    • No se admiten campos de secuencia.

    Uso recomendado: Pipelines de deduplicación donde solo importa el primer registro recibido.

    aggregation

    Paimon aplica una función de agregación a cada columna que no sea clave primaria en los registros que comparten las mismas claves primarias. Especifique la función por columna mediante fields.<field-name>.aggregate-function. Las columnas sin una función explícita usan last_non_null_value de forma predeterminada.

    CREATE TABLE T (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

    Resultado de ejemplo:

    Escriba +I(1, 23.0, 15) y +I(1, 30.2, 20) — la consulta devuelve (1, 30.2, 35).

    Funciones de agregación compatibles:

    Función

    Tipos compatibles

    sum

    DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE

    product

    DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE

    count

    INTEGER, BIGINT

    max, min

    CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ

    first_value, last_value

    Todos los tipos, incluido null

    first_not_null_value, last_non_null_value

    Todos los tipos

    listagg

    STRING

    bool_and, bool_or

    BOOLEAN
    Solo sum , product y count admiten cambios de retracción (UPDATE_BEFORE y DELETE). Para ignorar los cambios de retracción en una columna específica, establezca 'fields.<field-name>.ignore-retract' = 'true' .

    Establezca changelog-producer en lookup o full-compaction para habilitar el consumo mediante streaming.

    Uso recomendado: Métricas de agregación en tiempo real, totales acumulados, seguimiento de máximos/mínimos.

    partial-update

    Paimon actualiza columnas individuales de un registro existente utilizando el último valor no nulo de los registros entrantes que comparten las mismas claves primarias. Los valores nulos no sobrescriben los valores existentes.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 BIGINT,
  v3 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update'
);

    Resultado de ejemplo:

    Escriba +I(1, 23.0, 10, NULL), +I(1, NULL, NULL, 'This is a book'), +I(1, 25.2, NULL, NULL) — la consulta devuelve (1, 25.2, 10, 'This is a book').

    Notas de uso:

    • Establezca changelog-producer en lookup o full-compaction para habilitar el consumo mediante streaming.

    • El motor partial-update no procesa cambios de tipo DELETE ni UPDATE_BEFORE. Para ignorarlos, establezca 'partial-update.ignore-delete' = 'true'.

    Grupos de secuencia

    Al ensamblar una tabla ancha a partir de varias tablas de origen, utilice grupos de secuencia para controlar el orden de actualización de diferentes conjuntos de columnas de manera independiente, gestionando así los datos desordenados entre orígenes.

    En el siguiente ejemplo, las columnas a y b se actualizan en orden ascendente de g_1, y las columnas c y d se actualizan en orden ascendente de g_2:

    CREATE TABLE T (
  k INT,
  a STRING,
  b STRING,
  g_1 INT,
  c STRING,
  d STRING,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.g_2.sequence-group' = 'c,d'
);

    Combine grupos de secuencia con funciones de agregación añadiendo fields.<field-name>.aggregate-function para cualquier columna del grupo. El siguiente ejemplo rastrea el valor no nulo más reciente de a, el máximo de b, el valor no nulo más reciente de c y la suma de d:

    CREATE TABLE T (
  k INT,
  a STRING,
  b INT,
  g_1 INT,
  c STRING,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.b.aggregate-function' = 'max',
  'fields.g_2.sequence-group' = 'c,d',
  'fields.d.aggregate-function' = 'sum'
);

    Para obtener más información, consulte Merge Engine.

    Uso recomendado: Ensamblaje de tablas anchas mediante la unión de columnas de varios orígenes de streaming.

    Manejo de datos desordenados

    De forma predeterminada, Paimon fusiona los registros en el orden de entrada; el último registro recibido es el último en fusionarse. Para flujos con registros desordenados, establezca sequence.field en la cláusula WITH. Los registros con las mismas claves primarias se fusionarán entonces en orden ascendente según el valor de la columna especificada.

    Tipos de datos compatibles para sequence.field: TINYINT, SMALLINT, INTEGER, BIGINT, TIMESTAMP, TIMESTAMP_LTZ.

    Cuando se usa MySQL como origen de entrada con la columna de metadatos op_t como campo de secuencia, los pares de cambios UPDATE_BEFORE y UPDATE_AFTER comparten el mismo valor de secuencia. Para garantizar que Paimon procese UPDATE_BEFORE antes que UPDATE_AFTER, establezca 'sequence.auto-padding' = 'row-kind-flag' .

    Productor de changelog

    Una tabla de clave primaria debe generar un changelog completo (que cubra operaciones INSERT, DELETE y UPDATE) para admitir el consumo posterior mediante streaming, de manera similar al binlog de una base de datos. Configure el método de generación con el parámetro changelog-producer.

    Valor

    Descripción

    Latencia

    Uso de recursos

    Uso recomendado

    none

    No se genera changelog.

    —

    Mínimo

    No se requiere consumo mediante streaming

    input

    Pasa los registros de entrada directamente a los consumidores posteriores.

    Mínima

    Mínimo

    La entrada ya contiene un changelog completo, como el binlog de una base de datos. Es la opción más eficiente, ya que no requiere cómputo adicional.

    lookup (recomendado)

    Ejecuta una búsqueda contra los resultados de compactación de archivos pequeños para producir un changelog completo. Se activa en cada checkpoint de Flink.

    Nivel de minutos

    Mayor

    Cualquier tipo de entrada; requisitos de latencia a nivel de minutos

    full-compaction

    Genera un changelog después de cada compactación completa de archivos pequeños.

    Hasta varias horas

    Menor que lookup

    Cualquier tipo de entrada; se acepta una latencia mayor

    Para full-compaction, establezca 'full-compaction.delta-commits' = '<num>' para activar la compactación completa después de cada <num> checkpoints de Flink. La compactación completa consume muchos recursos; configure el intervalo entre 30 minutos y 1 hora.

    De forma predeterminada, Paimon genera un registro de changelog incluso cuando el valor actualizado es idéntico al anterior. Para suprimir estos registros sin cambios, establezca 'changelog-producer.row-deduplicate' = 'true' . Esta opción solo aplica a lookup y full-compaction , y requiere cómputo adicional para comparar los valores antes y después, así que habilítela solo cuando se espere una gran cantidad de registros innecesarios.

    Tablas de solo anexión

    Una tabla de solo anexión no tiene claves primarias y solo acepta operaciones INSERT en modo streaming. Utilice este tipo de tabla para cargas de trabajo que no requieren actualizaciones mediante streaming, como la sincronización de datos de registros.

    CREATE TABLE T (
  dt STRING,
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) PARTITIONED BY (dt) WITH (
  'bucket' = '-1'
);

    Subtipos

    Subtipo

    Configuración

    Características

    Tabla escalable de anexión

    'bucket' = '-1'

    Alto rendimiento de escritura; no requiere particionamiento por hash; admite ordenamiento de datos; configuración flexible de paralelismo; admite conversión directa desde tablas Hive; compactación de archivos completamente asíncrona; el orden de consumo difiere del orden de escritura; posible sesgo de datos cuando el paralelismo del origen equivale al paralelismo de escritura

    Tabla de cola de anexión

    'bucket' = '<num>' (entero > 0)

    Preserva el orden por bucket; latencia de varios minutos; equivalente a una partición de tema de Kafka o al número de shards en una instancia de ApsaraMQ for MQTT

    Asignación de buckets

    Tabla escalable de anexión: Los datos se escriben directamente en una sola partición en paralelo. Se ignora el concepto de bucket y no se requiere particionamiento por hash, lo que maximiza el rendimiento de escritura.

    Tabla de cola de anexión: De forma predeterminada, los registros se asignan a los buckets según los valores de todas las columnas del registro de datos. Para controlar la asignación, establezca bucket-key en la cláusula WITH con una lista de nombres de columna separados por comas.

    Por ejemplo, 'bucket-key' = 'c1,c2' enruta los registros según el hash de c1 y c2.

    Especifique bucket-key para reducir el cómputo de asignación de buckets y mejorar la eficiencia de escritura.

    Orden de consumo de datos

    Tabla escalable de anexión: Los registros pueden consumirse en un orden diferente al de escritura.

    Tabla de cola de anexión: Los registros dentro de cada bucket se consumen en el orden de escritura. Para registros en distintas particiones o buckets:

    • Dos registros de distintas particiones: si 'scan.plan-sort-partition' = 'true', se consume primero el registro con el menor valor de partición; de lo contrario, se consume primero el registro de la partición creada anteriormente.

    • Dos registros de la misma partición, mismo bucket: se consume primero el registro escrito anteriormente.

    • Dos registros de la misma partición, distintos buckets: no se garantiza el orden de consumo porque distintos buckets pueden procesarse simultáneamente.

    Pasos siguientes