Hudi connector - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Importante

El conector integrado de Hudi ya no será compatible en versiones futuras de Ververica Runtime (VVR). Utilice custom connectors para conectar Realtime Compute for Apache Flink con Apache Hudi, o migre a Paimon connector para obtener funciones y rendimiento optimizados.

Apache Hudi es un framework de data lake de código abierto que gestiona datos tabulares almacenados en Object Storage Service (OSS) o Hadoop Distributed File System (HDFS). Ofrece garantías ACID (Atomicidad, Consistencia, Aislamiento, Durabilidad), operaciones upsert y delete a nivel de fila, gestión automática de archivos pequeños y consultas de viaje en el tiempo.

Funciones principales

Función	Descripción
Semántica ACID	Aislamiento de instantáneas de forma predeterminada, lo que garantiza la consistencia de los datos durante lecturas y escrituras simultáneas.
Semántica UPSERT	Combina INSERT y UPDATE: si un registro no existe, se inserta; si ya existe, se actualiza. Esto simplifica el código de desarrollo ETL.
Viaje en el tiempo	Permite acceder a versiones históricas de datos en un momento específico, lo que facilita la auditoría eficiente de datos y el control de calidad.

Escenarios típicos

Escenario	Descripción
Aceleración de ingesta de bases de datos	Escriba datos de Change Data Capture (CDC) (por ejemplo, registros binarios de MySQL mediante el conector MySQL CDC) directamente en una tabla Hudi para procesos ETL en tiempo real posteriores; esta opción es más rentable que la carga masiva sin conexión.
ETL incremental	Extraiga flujos de datos de cambios de Hudi de forma incremental para realizar procesos ETL ligeros y en tiempo real. Utilice Apache Presto o Apache Spark para análisis OLAP posteriores.
Colas de mensajes	Utilice Hudi como reemplazo ligero de colas de mensajes para escenarios de bajo volumen, lo que simplifica la arquitectura de la aplicación.
Relleno de datos	Combine datos completos e incrementales de tablas Hudi en un metastore de Hive para generar tablas anchas con una sobrecarga de cómputo mínima.

Ventajas sobre Hudi de código abierto

Sin mantenimiento: El conector integrado de Hudi reduce la complejidad de operación y mantenimiento, y ofrece garantías de SLA.
Mayor conectividad de datos: Desacopla los datos de los motores de cómputo, lo que permite una migración fluida entre Apache Flink, Apache Spark, Apache Presto y Apache Hive.
Ingesta simplificada de base de datos a lago: Funciona con el conector Flink CDC para agilizar el desarrollo de datos.
Funciones de clase empresarial: Gestión unificada de metadatos mediante Data Lake Formation (DLF) y cambios de esquema ligeros automáticos.
Almacenamiento rentable: Los datos se almacenan en formato Apache Parquet o Apache Avro en Alibaba Cloud OSS, con aislamiento de almacenamiento y cómputo para un escalado flexible de recursos.

Configuraciones compatibles

Elemento	Valor
Tipo de tabla	Tabla de origen, tabla de destino
Modo de ejecución	Modo streaming, modo por lotes
Formato de datos	N/A
Tipo de API	DataStream API, SQL API
Actualización/eliminación de datos en destino	Compatible
Versión mínima de VVR	vvr-4.0.11-flink-1.13
Sistemas de archivos compatibles	OSS, HDFS, OSS-HDFS

Métricas

Tipo de tabla	Métricas
Tabla de origen	`numRecordsIn`, `numRecordsInPerSecond`
Tabla de destino	`numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`

Para consultar las definiciones de las métricas, véase Métricas.

Limitaciones

Versión mínima del motor: vvr-4.0.11-flink-1.13 o posterior.
Sistemas de archivos compatibles: únicamente OSS, HDFS u OSS-HDFS.
Los trabajos en borrador no pueden ejecutarse en clústeres de sesión.
El conector de Hudi no admite modificaciones de campos. Para modificar campos, utilice sentencias Spark SQL en la consola de Data Lake Formation (DLF).

Sintaxis

CREATE TEMPORARY TABLE hudi_tbl (
  uuid BIGINT,
  data STRING,
  ts   TIMESTAMP(3),
  PRIMARY KEY(uuid) NOT ENFORCED
) WITH (
  'connector' = 'hudi',
  'path' = 'oss://<yourOSSBucket>/<Custom storage directory>',
  ...
);

Parámetros en la cláusula WITH

Parámetros básicos

Parámetros comunes

Parámetro	Obligatorio	Predeterminado	Descripción
`connector`	Sí	—	Establézcalo en `hudi`.
`path`	Sí	—	Ruta de almacenamiento de la tabla. Formatos compatibles: OSS (`oss://<bucket>/<user-defined-dir>`), HDFS (`hdfs://<user-defined-dir>`), OSS-HDFS (`oss://<bucket>.<oss-hdfs-endpoint>/<user-defined-dir>`). Las rutas de OSS-HDFS requieren VVR 8.0.3 o posterior. Encuentre el endpoint de OSS-HDFS en la sección Port de la página Overview del bucket de OSS.
`hoodie.datasource.write.recordkey.field`	No	`uuid`	Campo de clave primaria. Separe varios campos con comas. Como alternativa, utilice la sintaxis `PRIMARY KEY` en el DDL.
`precombine.field`	No	`ts`	Campo de versión utilizado para determinar el orden de las actualizaciones. Si no se establece, las actualizaciones siguen la secuencia de mensajes definida por el motor.
`oss.endpoint`	No	—	Obligatorio cuando se almacenan datos en OSS u OSS-HDFS. Para endpoints de OSS, consulte Regions and endpoints. Para endpoints de OSS-HDFS, consulte la sección Port de la página Overview del bucket de OSS.
`accessKeyId`	No	—	AccessKey ID. Obligatorio para OSS y OSS-HDFS. Almacene las credenciales como variables en lugar de codificarlas directamente. Consulte Manage variables.
`accessKeySecret`	No	—	AccessKey secret. Obligatorio para OSS y OSS-HDFS.

Importante

Para proteger su par de AccessKey, almacene el AccessKey ID y el AccessKey secret como variables. Consulte Gestionar variables.

Parámetros de tabla de origen

Parámetro	Obligatorio	Predeterminado	Descripción
`read.streaming.enabled`	No	`false`	Establézcalo en `true` para habilitar la lectura en streaming. De forma predeterminada, se utiliza la lectura de instantáneas, que devuelve la última instantánea completa.
`read.start-commit`	No	(vacío)	Desplazamiento inicial para la lectura en streaming. Formato: `yyyyMMddHHmmss` para una hora específica, o `earliest` para leer desde el inicio. Déjelo en blanco para leer desde el último commit.

Parámetros de tabla de destino

Parámetro	Obligatorio	Predeterminado	Descripción
`write.operation`	No	`UPSERT`	Modo de escritura. Valores válidos: `insert` (anexar), `upsert` (insertar o actualizar), `bulk_insert` (anexar por lotes).
`hive_sync.enable`	No	`false`	Establézcalo en `true` para sincronizar metadatos con Apache Hive.
`hive_sync.mode`	No	`hms`	Modo de sincronización. `hms` sincroniza con un metastore de Hive o DLF. `jdbc` sincroniza mediante un controlador Java Database Connectivity (JDBC).
`hive_sync.db`	No	`default`	Nombre de la base de datos Hive de destino.
`hive_sync.table`	No	Nombre de la tabla actual	Nombre de la tabla Hive de destino. No debe contener guiones (`-`).
`dlf.catalog.region`	No	—	Región donde está activado DLF. Solo surte efecto cuando `hive_sync.mode` es `hms`. Consulte Supported regions and endpoints. Debe coincidir con la región especificada por `dlf.catalog.endpoint`.
`dlf.catalog.endpoint`	No	—	Endpoint de DLF. Solo surte efecto cuando `hive_sync.mode` es `hms`. Utilice el endpoint de VPC para reducir la latencia; por ejemplo, `dlf-vpc.cn-hangzhou.aliyuncs.com` para la región China (Hangzhou). Consulte Supported regions and endpoints. Para acceso entre VPC, consulte How does Realtime Compute for Apache Flink access a service across VPCs?

Parámetros avanzados

Parámetros de paralelismo

Parámetro	Predeterminado	Descripción
`write.tasks`	`4`	Paralelismo de las tareas de escritura. Cada tarea escribe secuencialmente en uno o más buckets. Aumentar este valor no incrementa la cantidad de archivos pequeños.
`write.bucket_assign.tasks`	Paralelismo del despliegue	Paralelismo de los operadores de asignación de buckets. Aumentar este valor incrementa la cantidad de archivos pequeños.
`write.index_bootstrap.tasks`	Paralelismo del despliegue	Paralelismo de los operadores de arranque de índice. Solo surte efecto cuando `index.bootstrap.enabled` es `true`. Aumentar este valor mejora el rendimiento del arranque, pero puede bloquear los puntos de control durante el proceso; aumente la tolerancia a fallos de puntos de control si es necesario.
`read.tasks`	`4`	Paralelismo de los operadores de lectura en streaming y por lotes.
`compaction.tasks`	`4`	Paralelismo de los operadores de compactación en línea. La compactación en línea consume más recursos que la compactación sin conexión; prefiera la compactación sin conexión para cargas de trabajo de producción.

Parámetros de compactación en línea

Parámetro	Predeterminado	Descripción
`compaction.schedule.enabled`	`true`	Indica si se generan planes de compactación según una programación. Mantenga este valor en `true` incluso si la compactación asíncrona está deshabilitada; así, la compactación sin conexión podrá ejecutar los planes programados.
`compaction.async.enabled`	`true`	Indica si la compactación se ejecuta de forma asíncrona. Establézcalo en `false` para deshabilitar la compactación en línea manteniendo activa la generación de planes.
`compaction.tasks`	`4`	Paralelismo de las tareas de compactación.
`compaction.trigger.strategy`	`num_commits`	Estrategia utilizada para activar la compactación. Valores válidos: `num_commits`, `time_elapsed`, `num_and_time`, `num_or_time`.
`compaction.delta_commits`	`5`	Número de commits necesarios para activar la compactación. Se usa con `num_commits`, `num_and_time` o `num_or_time`.
`compaction.delta_seconds`	`3600`	Intervalo en segundos entre activaciones de compactación. Se usa con `time_elapsed`, `num_and_time` o `num_or_time`.
`compaction.max_memory`	`100 MB`	Memoria máxima para el mapa hash utilizado durante la compactación y la deduplicación. Auméntela a 1 GB si los recursos lo permiten.
`compaction.target_io`	`500 GB`	Rendimiento máximo de E/S por plan de compactación.

Parámetros de tamaño de archivo

Estos parámetros controlan cómo Hudi gestiona los tamaños de archivo para evitar la acumulación de archivos pequeños.

Parámetro	Predeterminado	Descripción
`hoodie.parquet.max.file.size`	120 MB (120 × 1024 × 1024 bytes)	Tamaño máximo de un archivo Parquet. Los datos que superen este umbral se escriben en un nuevo grupo de archivos.
`hoodie.parquet.small.file.limit`	100 MB (104,857,600 bytes)	Los archivos menores a este umbral se consideran archivos pequeños. Durante la escritura, Hudi anexa datos a los archivos pequeños existentes en lugar de crear nuevos.
`hoodie.copyonwrite.record.size.estimate`	1 KB (1,024 bytes)	Tamaño estimado del registro. Si no se establece, Hudi calcula este valor dinámicamente a partir de los metadatos confirmados.

Parámetros de configuración de Hadoop

Parámetro	Predeterminado	Descripción
`hadoop.${option key}`	—	Elementos de configuración de Hadoop, especificados con el prefijo `hadoop.`. Compatible con Hudi 0.12.0 y posteriores. Utilice sentencias DDL para especificar configuraciones de Hadoop por trabajo en escenarios entre clústeres. Se pueden especificar varios elementos simultáneamente.

Parámetros de escritura de datos

Escritura por lotes

Utilice la escritura por lotes para importar datos existentes de otras fuentes a una tabla Hudi.

bulk_insert omite la serialización Avro, la compactación y la deduplicación. Garantice la unicidad de los datos de origen antes de usar este modo. bulk_insert solo es válido en el modo de ejecución por lotes.

Parámetro	Predeterminado	Descripción
`write.operation`	`upsert`	Tipo de escritura. Establézcalo en `bulk_insert` para escrituras por lotes.
`write.tasks`	Paralelismo del despliegue	Paralelismo para las tareas de `bulk_insert`. El número final de archivos de salida es mayor o igual a este valor (los datos pasan a un nuevo archivo cuando se alcanza el límite de 120 MB de Parquet).
`write.bulk_insert.shuffle_input`	`true`	Indica si se reorganizan los datos de entrada por campo de partición antes de escribir. Disponible en Hudi 0.11.0 y posteriores. Reduce la cantidad de archivos pequeños, pero puede causar sesgo de datos.
`write.bulk_insert.sort_input`	`true`	Indica si se ordenan los datos de entrada por campo de partición antes de escribir. Disponible en Hudi 0.11.0 y posteriores. Reduce la cantidad de archivos pequeños cuando una sola tarea escribe en varias particiones.
`write.sort.memory`	`128`	Memoria gestionada disponible para el operador de ordenamiento, en MB.

Modo changelog

En el modo changelog, Hudi conserva todos los eventos de cambio (INSERT, UPDATE_BEFORE, UPDATE_AFTER y DELETE), lo que permite implementar almacenamiento de datos casi en tiempo real de extremo a extremo con la computación con estado de Flink. Las tablas Merge On Read (MOR) admiten este modo.

En el modo sin changelog, los cambios intermedios dentro de un lote se fusionan. La lectura de instantáneas devuelve solo el resultado final fusionado; los estados intermedios no son visibles independientemente de la ruta de escritura.

Después de habilitar el modo changelog, la tarea de compactación asíncrona sigue fusionando los cambios intermedios. Establezca compaction.delta_commits=5 y compaction.delta_seconds=3600 para dar a los consumidores posteriores tiempo suficiente para leer los registros antes de que se compacten.

Parámetro	Predeterminado	Descripción
`changelog.enabled`	`false`	Establézcalo en `true` para conservar todos los eventos de cambio. Cuando es `false`, solo se garantiza el registro final fusionado; los cambios intermedios pueden fusionarse.

Modo anexar

Compatible con Hudi 0.10.0 y posteriores.

Tablas MOR: Se aplica la política de archivos pequeños. Los datos se escriben en archivos de registro Apache Avro en modo anexar.
Tablas Copy On Write (COW): No se aplica la política de archivos pequeños. Se crea un nuevo archivo Apache Parquet para cada escritura.

Parámetros de clustering

Hudi admite clustering para resolver la acumulación de archivos pequeños en modo INSERT.

Clustering en línea (solo tablas COW)

Parámetro	Predeterminado	Descripción
`write.insert.cluster`	`false`	Establézcalo en `true` para fusionar archivos pequeños durante la escritura. Cada operación INSERT fusiona los archivos pequeños existentes, pero no se realiza deduplicación y el rendimiento de escritura disminuye.

Clustering asíncrono (Hudi 0.12.0 y posteriores)

Parámetro	Predeterminado	Descripción
`clustering.schedule.enabled`	`false`	Establézcalo en `true` para programar periódicamente un plan de clustering.
`clustering.delta_commits`	`4`	Número de commits necesarios para generar un plan de clustering. Solo surte efecto cuando `clustering.schedule.enabled` es `true`.
`clustering.async.enabled`	`false`	Establézcalo en `true` para ejecutar el plan de clustering de forma asíncrona a intervalos regulares.
`clustering.tasks`	`4`	Paralelismo de las tareas de clustering.
`clustering.plan.strategy.target.file.max.bytes`	1 GiB (1,073,741,824 bytes)	Tamaño máximo de archivo objetivo para la salida del clustering.
`clustering.plan.strategy.small.file.limit`	`600`	Los archivos menores a este umbral (en bytes) son elegibles para clustering.
`clustering.plan.strategy.sort.columns`	—	Columnas utilizadas para ordenar datos durante el clustering.

Estrategias de plan de clustering

Parámetro	Predeterminado	Descripción
`clustering.plan.partition.filter.mode`	`NONE`	Modo de filtro de partición. Valores válidos: `NONE` (todas las particiones), `RECENT_DAYS` (particiones de los últimos N días), `SELECTED_PARTITIONS` (particiones específicas).
`clustering.plan.strategy.daybased.lookback.partitions`	`2`	Número de días recientes para seleccionar particiones para clustering. Solo surte efecto cuando `filter.mode` es `RECENT_DAYS`.
`clustering.plan.strategy.cluster.begin.partition`	—	Partición inicial para filtrado por rango. Solo surte efecto cuando `filter.mode` es `SELECTED_PARTITIONS`.
`clustering.plan.strategy.cluster.end.partition`	—	Partición final para filtrado por rango. Solo surte efecto cuando `filter.mode` es `SELECTED_PARTITIONS`.
`clustering.plan.strategy.partition.regex.pattern`	—	Expresión regular para seleccionar particiones.
`clustering.plan.strategy.partition.selected`	—	Lista separada por comas de particiones seleccionadas.

Elegir un tipo de índice

Hudi admite dos tipos de índice. Utilice esta tabla para elegir el que mejor se adapte a su carga de trabajo.

Dimensión	FLINK_STATE	BUCKET
Sobrecarga de almacenamiento/cómputo	Sí (backend de estado)	Ninguna
Rendimiento	Depende del backend de estado	Mejor (sin sobrecarga de estado)
Flexibilidad de grupos de archivos	Asigna registros dinámicamente según el tamaño del archivo	Número fijo de buckets (no se puede aumentar después de la configuración inicial)
Cambios entre particiones	Compatible	No compatible (excepción: entrada streaming de Change Data Capture (CDC))
Cuándo utilizarlo	Tablas con menos de 500 millones de registros, o cargas de trabajo que requieren actualizaciones entre particiones	Tablas con más de 500 millones de registros donde la sobrecarga de estado es un cuello de botella

Cuando index.type se establece en BUCKET , establecer index.global.enabled=true no tiene efecto, ya que el índice de bucket no admite deduplicación entre particiones.

Parámetro	Predeterminado	Descripción
`index.type`	`FLINK_STATE`	Tipo de índice. Valores válidos: `FLINK_STATE`, `BUCKET`.
`hoodie.bucket.index.hash.field`	Clave primaria	Campo de clave hash para el índice de bucket. Puede ser un subconjunto de la clave primaria.
`hoodie.bucket.index.num.buckets`	`4`	Número de buckets por partición. No se puede cambiar después de crear la tabla.

Los parámetros de índice de bucket son compatibles con Hudi 0.11.0 y posteriores.

Parámetros de lectura de datos

Hudi admite tres patrones de lectura utilizando el mismo conjunto de parámetros.

Patrón	Configuración
Lectura en streaming	Establezca `read.streaming.enabled=true` y opcionalmente `read.start-commit`
Lectura incremental por lotes	Establezca tanto `read.start-commit` como `read.end-commit`; el intervalo es cerrado (inclusivo en ambos extremos)
Viaje en el tiempo	Establezca solo `read.end-commit`; lee una instantánea en ese commit específico

Parámetros de lectura en streaming

De forma predeterminada, la lectura de una tabla Hudi utiliza lectura de instantáneas, devolviendo la última instantánea completa de una vez. Establezca read.streaming.enabled=true para cambiar a lectura en streaming.

Parámetro	Predeterminado	Descripción
`read.streaming.enabled`	`false`	Establézcalo en `true` para habilitar la lectura en streaming.
`read.start-commit`	(vacío)	Desplazamiento inicial. Formato: `yyyyMMddHHmmss` para una hora específica, o `earliest` para leer desde el inicio. Déjelo en blanco para comenzar desde el último commit.
`clean.retain_commits`	`30`	Número máximo de commits históricos retenidos por el limpiador. Los commits que superen este límite se eliminan. Por ejemplo, con un intervalo de puntos de control de 5 minutos, el valor predeterminado de 30 retiene los changelogs durante al menos 150 minutos.

Importante

La lectura en streaming de changelogs requiere Hudi 0.10.0 o posterior. Las tareas de compactación pueden fusionar changelogs, eliminando registros intermedios y afectando potencialmente los cálculos posteriores.

Parámetros de lectura incremental

Parámetro	Predeterminado	Descripción
`read.start-commit`	Último commit	Inicio del rango de lectura, en formato `yyyyMMddHHmmss`.
`read.end-commit`	Último commit	Fin del rango de lectura, en formato `yyyyMMddHHmmss`. El rango es cerrado (ambos extremos inclusivos).

Ejemplos

Tabla de origen

CREATE TEMPORARY TABLE blackhole (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'blackhole'
);

CREATE TEMPORARY TABLE hudi_tbl (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'hudi',
  'oss.endpoint' = '<yourOSSEndpoint>',
  'accessKeyId' = '${secret_values.ak_id}',
  'accessKeySecret' = '${secret_values.ak_secret}',
  'path' = 'oss://<yourOSSBucket>/<Custom storage directory>',
  'table.type' = 'MERGE_ON_READ',
  'read.streaming.enabled' = 'true'
);

-- Read from the latest commit in streaming mode and write to Blackhole.
INSERT INTO blackhole SELECT * FROM hudi_tbl;

Tabla de destino

CREATE TEMPORARY TABLE datagen (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '100'
);

CREATE TEMPORARY TABLE hudi_tbl (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'hudi',
  'oss.endpoint' = '<yourOSSEndpoint>',
  'accessKeyId' = '${secret_values.ak_id}',
  'accessKeySecret' = '${secret_values.ak_secret}',
  'path' = 'oss://<yourOSSBucket>/<Custom storage directory>',
  'table.type' = 'MERGE_ON_READ'
);

INSERT INTO hudi_tbl SELECT * FROM datagen;

DataStream API

Importante

Para utilizar DataStream API, configure un conector DataStream para Realtime Compute for Apache Flink. Consulte Settings of DataStream connectors.

Dependencias de Maven

Ajuste las versiones de las dependencias a su versión de VVR.

<properties>
  <maven.compiler.source>8</maven.compiler.source>
  <maven.compiler.target>8</maven.compiler.target>
  <flink.version>1.15.4</flink.version>
  <hudi.version>0.13.1</hudi.version>
</properties>

<dependencies>
  <!-- Flink -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>

  <!-- Hudi -->
  <dependency>
    <groupId>org.apache.hudi</groupId>
    <artifactId>hudi-flink1.15-bundle</artifactId>
    <version>${hudi.version}</version>
    <scope>provided</scope>
  </dependency>

  <!-- OSS -->
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.3.2</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-aliyun</artifactId>
    <version>3.3.2</version>
    <scope>provided</scope>
  </dependency>

  <!-- DLF -->
  <dependency>
    <groupId>com.aliyun.datalake</groupId>
    <artifactId>metastore-client-hive2</artifactId>
    <version>0.2.14</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>2.5.1</version>
    <scope>provided</scope>
  </dependency>
</dependencies>

Importante

Las dependencias de DLF entran en conflicto con las versiones de código abierto de Apache Hive (hive-common, hive-exec). Para pruebas locales de DLF, descargue los paquetes JAR personalizados hive-common y hive-exec e impórtelos manualmente en IntelliJ IDEA.

Escribir datos en Hudi

El siguiente ejemplo escribe datos en una tabla Hudi MOR en OSS y, opcionalmente, sincroniza metadatos con DLF.

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.util.HoodiePipeline;

import java.util.HashMap;
import java.util.Map;

public class FlinkHudiQuickStart {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    String dbName = "test_db";
    String tableName = "test_tbl";
    String basePath = "oss://xxx";

    Map<String, String> options = new HashMap<>();

    // Hudi configuration
    options.put(FlinkOptions.PATH.key(), basePath);
    options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
    options.put(FlinkOptions.PRECOMBINE_FIELD.key(), "ts");
    options.put(FlinkOptions.DATABASE_NAME.key(), dbName);
    options.put(FlinkOptions.TABLE_NAME.key(), tableName);

    // OSS configuration
    // Use the public endpoint for local debugging (e.g., oss-cn-hangzhou.aliyuncs.com)
    // Use the internal endpoint for cluster submission (e.g., oss-cn-hangzhou-internal.aliyuncs.com)
    options.put("hadoop.fs.oss.accessKeyId", "xxx");
    options.put("hadoop.fs.oss.accessKeySecret", "xxx");
    options.put("hadoop.fs.oss.endpoint", "xxx");
    options.put("hadoop.fs.AbstractFileSystem.oss.impl", "org.apache.hadoop.fs.aliyun.oss.OSS");
    options.put("hadoop.fs.oss.impl", "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem");

    // DLF configuration (optional — remove if not syncing to DLF)
    // Use the public endpoint for local debugging (e.g., dlf.cn-hangzhou.aliyuncs.com)
    // Use the VPC endpoint for cluster submission (e.g., dlf-vpc.cn-hangzhou.aliyuncs.com)
    options.put(FlinkOptions.HIVE_SYNC_ENABLED.key(), "true");
    options.put(FlinkOptions.HIVE_SYNC_MODE.key(), "hms");
    options.put(FlinkOptions.HIVE_SYNC_DB.key(), dbName);
    options.put(FlinkOptions.HIVE_SYNC_TABLE.key(), tableName);
    options.put("hadoop.dlf.catalog.id", "xxx");
    options.put("hadoop.dlf.catalog.accessKeyId", "xxx");
    options.put("hadoop.dlf.catalog.accessKeySecret", "xxx");
    options.put("hadoop.dlf.catalog.region", "xxx");
    options.put("hadoop.dlf.catalog.endpoint", "xxx");
    options.put("hadoop.hive.imetastoreclient.factory.class",
        "com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory");

    DataStream<RowData> dataStream = env.fromElements(
        GenericRowData.of(StringData.fromString("id1"), StringData.fromString("name1"), 22,
            StringData.fromString("1001"), StringData.fromString("p1")),
        GenericRowData.of(StringData.fromString("id2"), StringData.fromString("name2"), 32,
            StringData.fromString("1002"), StringData.fromString("p2"))
    );

    HoodiePipeline.Builder builder = HoodiePipeline.builder(tableName)
        .column("uuid string")
        .column("name string")
        .column("age int")
        .column("ts string")
        .column("`partition` string")
        .pk("uuid")
        .partition("partition")
        .options(options);

    // Second parameter: whether the input stream is bounded (true = batch, false = streaming)
    builder.sink(dataStream, false);
    env.execute("Flink_Hudi_Quick_Start");
  }
}

Realtime Compute for Apache Flink:Hudi (retiring)

Funciones principales

Escenarios típicos

Ventajas sobre Hudi de código abierto

Configuraciones compatibles

Limitaciones

Sintaxis

Parámetros en la cláusula WITH

Parámetros básicos

Parámetros avanzados

Parámetros de paralelismo

Parámetros de compactación en línea

Parámetros de tamaño de archivo

Parámetros de configuración de Hadoop

Parámetros de escritura de datos

Parámetros de clustering

Elegir un tipo de índice

Parámetros de lectura de datos

Ejemplos

Tabla de origen

Tabla de destino

DataStream API

Dependencias de Maven

Escribir datos en Hudi

Preguntas frecuentes