diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index ce9fca5e2b..d9d79502d8 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1798,8 +1798,8 @@ public class ConfigOptions { .enumType(DataLakeFormat.class) .noDefaultValue() .withDescription( - "The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, and `lance`. " - + "In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi. " + "The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, `hudi`, and `lance`. " + + "In the future, more kinds of data lake formats will be supported, such as DeltaLake. " + "Once the `table.datalake.format` property is configured, Fluss adopts the key encoding and bucketing strategy used by the corresponding data lake format. " + "This ensures consistency in key encoding and bucketing, enabling seamless **Union Read** functionality across Fluss and Lakehouse. " + "The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. " @@ -2415,8 +2415,8 @@ public class ConfigOptions { .enumType(DataLakeFormat.class) .noDefaultValue() .withDescription( - "The datalake format used by Fluss as lakehouse storage. Currently, supported formats are Paimon, Iceberg, and Lance. " - + "In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi."); + "The datalake format used by Fluss as lakehouse storage. Currently, supported formats are Paimon, Iceberg, Hudi, and Lance. " + + "In the future, more kinds of data lake format will be supported, such as DeltaLake."); // ------------------------------------------------------------------------ // ConfigOptions for tiering service diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index c612a968f6..4b131252d2 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -84,7 +84,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) | table.kv.standby-replica.enabled | Boolean | (None) | Whether to enable standby replicas for primary key tables. Standby replicas maintain recent KV snapshots for fast leader promotion. Automatically set to `true` by the coordinator during table creation for new PK tables. Tables created before this option was introduced are treated as disabled. Can be dynamically enabled via `ALTER TABLE SET ('table.kv.standby-replica.enabled' = 'true')`. | | table.log.tiered.local-segments | Integer | 2 | The number of log segments to retain in local for each table when log tiered storage is enabled. It must be greater that 0. The default is 2. | | table.datalake.enabled | Boolean | false | Whether enable lakehouse storage for the table. Disabled by default. When this option is set to ture and the datalake tiering service is up, the table will be tiered and compacted into datalake format stored on lakehouse storage. | -| table.datalake.format | Enum | (None) | The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, and `lance`. In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi. Once the `table.datalake.format` property is configured, Fluss adopts the key encoding and bucketing strategy used by the corresponding data lake format. This ensures consistency in key encoding and bucketing, enabling seamless **Union Read** functionality across Fluss and Lakehouse. The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. If `table.datalake.format` is not explicitly set during table creation, the table will default to the format specified by the `datalake.format` configuration in the Fluss cluster. | +| table.datalake.format | Enum | (None) | The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, `hudi`, and `lance`. In the future, more kinds of data lake format will be supported, such as DeltaLake. Once the `table.datalake.format` property is configured, Fluss adopts the key encoding and bucketing strategy used by the corresponding data lake format. This ensures consistency in key encoding and bucketing, enabling seamless **Union Read** functionality across Fluss and Lakehouse. The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. If `table.datalake.format` is not explicitly set during table creation, the table will default to the format specified by the `datalake.format` configuration in the Fluss cluster. | | table.datalake.freshness | Duration | 3min | It defines the maximum amount of time that the datalake table's content should lag behind updates to the Fluss table. Based on this target freshness, the Fluss service automatically moves data from the Fluss table and updates to the datalake table, so that the data in the datalake table is kept up to date within this target. If the data does not need to be as fresh, you can specify a longer target freshness time to reduce costs. | | table.datalake.auto-compaction | Boolean | false | If true, compaction will be triggered automatically when tiering service writes to the datalake. It is disabled by default. | | table.datalake.auto-expire-snapshot | Boolean | false | If true, snapshot expiration will be triggered automatically when tiering service commits to the datalake. It is disabled by default. | diff --git a/website/docs/engine-spark/reads.md b/website/docs/engine-spark/reads.md index c9d745e45d..ca48bdb201 100644 --- a/website/docs/engine-spark/reads.md +++ b/website/docs/engine-spark/reads.md @@ -202,7 +202,7 @@ ORDER BY order_id; When a table has the configuration `table.datalake.enabled = 'true'`, its data exists in two layers: - **Fresh data** is retained in Fluss (real-time layer, sub-second freshness) -- **Historical data** is tiered to the lake storage (e.g., Paimon, Iceberg) +- **Historical data** is tiered to the lake storage (e.g., Paimon, Iceberg, Hudi) Fluss Spark connector supports **union read** that combines both layers to provide a complete, up-to-date view of the data. This allows Fluss to store only a small portion of the dataset in the cluster (reducing costs), while the lake serves as the source of complete historical data. diff --git a/website/docs/install-deploy/deploying-streaming-lakehouse.md b/website/docs/install-deploy/deploying-streaming-lakehouse.md index 370adf909c..e3752fc6aa 100644 --- a/website/docs/install-deploy/deploying-streaming-lakehouse.md +++ b/website/docs/install-deploy/deploying-streaming-lakehouse.md @@ -29,7 +29,7 @@ Configure lakehouse settings in `server.yaml` on all Fluss servers (CoordinatorS Fluss follows a simple convention for these keys: - `datalake.enabled: true` explicitly enables lakehouse capability for the cluster. If it is left unset, configuring `datalake.format` alone also enables it; whenever `datalake.enabled` is `true`, `datalake.format` must also be set. -- `datalake.format` selects the lake format for the cluster (`paimon`, `iceberg`, or `lance`). +- `datalake.format` selects the lake format for the cluster (`paimon`, `iceberg`, `hudi`, or `lance`). - Format-specific options use the `datalake..*` prefix. Fluss strips this prefix and passes the remaining keys straight to the corresponding lake catalog/client — for example `datalake.paimon.metastore` becomes `metastore` and `datalake.iceberg.type` becomes `type`. Any option supported by the lake catalog can be set this way. - Only configure options for the format selected by `datalake.format`. @@ -71,6 +71,25 @@ datalake.iceberg.warehouse: s3://bucket/iceberg datalake.iceberg.io-impl: org.apache.iceberg.aws.s3.S3FileIO ``` + + + +```yaml title="server.yaml" +datalake.enabled: true +datalake.format: hudi +datalake.hudi.mode: dfs +datalake.hudi.catalog.path: /path/to/hudi +``` + +For Hive Metastore catalog: +```yaml title="server.yaml" +datalake.enabled: true +datalake.format: hudi +datalake.hudi.mode: hms +datalake.hudi.catalog.path: hdfs:///warehouse/hudi +datalake.hudi.hadoop.hive.metastore.uris: thrift://:9083 +``` + @@ -122,6 +141,16 @@ Add JARs to `${FLUSS_HOME}/plugins//` based on your configuration: | Iceberg with S3 | [`iceberg-aws-.jar`, `iceberg-aws-bundle-.jar`](https://iceberg.apache.org/docs/1.10.1/aws/), matching your Iceberg version | | Iceberg JDBC catalog | PostgreSQL/MySQL JDBC driver | + + + +| Scenario | Required JAR | +|----------|--------------| +| Hudi lake connector | [fluss-lake-hudi-$FLUSS_VERSION$.jar](https://repo1.maven.org/maven2/org/apache/fluss/fluss-lake-hudi/$FLUSS_VERSION$/fluss-lake-hudi-$FLUSS_VERSION$.jar) | +| Hudi Flink bundle | [hudi-flink1.20-bundle-1.1.0.jar](https://repo.maven.apache.org/maven2/org/apache/hudi/hudi-flink1.20-bundle/1.1.0/hudi-flink1.20-bundle-1.1.0.jar), matching the Flink version used by the Hudi integration | +| Hudi Flink dependencies | `flink-core`, `flink-table-common`, `flink-table-api-java`, and `flink-table-runtime`, matching the Hudi Flink bundle | +| Hudi Hive Metastore catalog | Hive and Hadoop dependencies required by your environment | + @@ -160,6 +189,14 @@ Add the following to `${FLINK_HOME}/lib`: - Hadoop client JARs — export `HADOOP_CLASSPATH`, download the pre-bundled [`hadoop-apache-3.3.5-2.jar`](https://repo1.maven.org/maven2/io/trino/hadoop/hadoop-apache/3.3.5-2/hadoop-apache-3.3.5-2.jar), or install a full Hadoop package (see [Iceberg Hadoop Dependencies](../streaming-lakehouse/datalake-formats/iceberg.md#prerequisites-hadoop-dependencies)) - JDBC driver (if using JDBC catalog) + + + +- [fluss-flink-1.20-$FLUSS_VERSION$.jar](https://repo1.maven.org/maven2/org/apache/fluss/fluss-flink-1.20/$FLUSS_VERSION$/fluss-flink-1.20-$FLUSS_VERSION$.jar) +- [fluss-lake-hudi-$FLUSS_VERSION$.jar](https://repo1.maven.org/maven2/org/apache/fluss/fluss-lake-hudi/$FLUSS_VERSION$/fluss-lake-hudi-$FLUSS_VERSION$.jar) +- [hudi-flink1.20-bundle-1.1.0.jar](https://repo.maven.apache.org/maven2/org/apache/hudi/hudi-flink1.20-bundle/1.1.0/hudi-flink1.20-bundle-1.1.0.jar), matching your Flink runtime +- Hadoop, Hive, or filesystem dependencies required by your Hudi catalog and table storage + @@ -199,6 +236,17 @@ ${FLINK_HOME}/bin/flink run /path/to/fluss-flink-tiering-$FLUSS_VERSION$.jar \ --datalake.iceberg.warehouse "s3://bucket/iceberg" ``` + + + +```shell +${FLINK_HOME}/bin/flink run /path/to/fluss-flink-tiering-$FLUSS_VERSION$.jar \ + --fluss.bootstrap.servers localhost:9123 \ + --datalake.format hudi \ + --datalake.hudi.mode dfs \ + --datalake.hudi.catalog.path /tmp/hudi +``` + @@ -237,14 +285,16 @@ CREATE TABLE my_table ( ## Verification 1. Check the Tiering Service job is running in Flink Web UI -2. After the freshness interval, query the lake table: +2. After the freshness interval, query the table: ```sql title="Flink SQL" --- Lake-only query -SELECT * FROM my_table$lake; - -- Union Read (real-time + historical) SELECT * FROM my_table; + +-- Lake-only query for formats exposed through the Fluss lake catalog +SELECT * FROM my_table$lake; ``` +For Hudi lake-only reads, use Hudi's native catalog or another Hudi-compatible engine. The Fluss catalog `$lake` suffix currently exposes Paimon and Iceberg lake tables. + See [Union Read](../streaming-lakehouse/union-read.md) for more details. diff --git a/website/docs/install-deploy/overview.mdx b/website/docs/install-deploy/overview.mdx index cf325d2d91..bf90355892 100644 --- a/website/docs/install-deploy/overview.mdx +++ b/website/docs/install-deploy/overview.mdx @@ -121,6 +121,7 @@ We have listed them in the table below the figure.
  • [Paimon](../streaming-lakehouse/datalake-formats/paimon.md)
  • [Iceberg](../streaming-lakehouse/datalake-formats/iceberg.md)
    • +
  • [Hudi](../streaming-lakehouse/datalake-formats/hudi.md)
  • [Lance](../streaming-lakehouse/datalake-formats/lance.md)
    • @@ -147,4 +148,4 @@ Fluss can be deployed in three different ways: - [Docker run / Docker Compose](install-deploy/deploying-with-docker.md) **NOTE**: -- Local Cluster is for testing purpose only. \ No newline at end of file +- Local Cluster is for testing purpose only. diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index d959b4eed6..52b58f3f54 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -225,7 +225,7 @@ More metrics example could be found in [Observability - Metric Reporters](observ | Option | Type | Default | Description | |------------------|---------|---------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | datalake.enabled | Boolean | (None) | Whether the Fluss cluster is ready to create and manage lakehouse tables. If unset, Fluss keeps the legacy behavior where configuring `datalake.format` also enables lakehouse tables. If set to `false`, Fluss pre-binds the lake format for newly created tables but does not allow lakehouse tables yet. If set to `true`, Fluss fully enables lakehouse tables. When this option is explicitly configured to true, `datalake.format` must also be configured. | -| datalake.format | Enum | (None) | The datalake format used by Fluss as lakehouse storage. Currently, supported formats are Paimon, Iceberg, and Lance. In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi. | +| datalake.format | Enum | (None) | The datalake format used by Fluss as lakehouse storage. Currently, supported formats are Paimon, Iceberg, Hudi, and Lance. In the future, more kinds of data lake format will be supported, such as DeltaLake. | ## Kafka diff --git a/website/docs/maintenance/tiered-storage/lakehouse-storage.md b/website/docs/maintenance/tiered-storage/lakehouse-storage.md index a9174073dd..4c909b94ef 100644 --- a/website/docs/maintenance/tiered-storage/lakehouse-storage.md +++ b/website/docs/maintenance/tiered-storage/lakehouse-storage.md @@ -5,7 +5,7 @@ sidebar_position: 3 # Lakehouse Storage -Fluss leverages well-known Lakehouse storage solutions like Apache Paimon, Apache Iceberg, and Lance as the tiered storage layer. The Tiering Service continuously tiers Fluss data to Lakehouse storage, where it can be read by Fluss clients in a streaming manner and accessed directly by external systems such as Flink, Spark, StarRocks, and others. +Fluss leverages well-known Lakehouse storage solutions like Apache Paimon, Apache Iceberg, Apache Hudi, and Lance as the tiered storage layer. The Tiering Service continuously tiers Fluss data to Lakehouse storage, where it can be read by Fluss clients in a streaming manner and accessed directly by external systems such as Flink, Spark, StarRocks, and others. For deployment instructions, see [Deploying Streaming Lakehouse](../../install-deploy/deploying-streaming-lakehouse.md). @@ -46,6 +46,7 @@ Fluss processes configurations by removing the `datalake..` prefix and u For format-specific configuration, see: - [Paimon](../../streaming-lakehouse/datalake-formats/paimon.md) - [Iceberg](../../streaming-lakehouse/datalake-formats/iceberg.md) +- [Hudi](../../streaming-lakehouse/datalake-formats/hudi.md) - [Lance](../../streaming-lakehouse/datalake-formats/lance.md) ## Table-Level Options @@ -54,7 +55,7 @@ For format-specific configuration, see: |--------|------|---------|-------------| | `table.datalake.enabled` | Boolean | false | Enable lakehouse storage for this table | | `table.datalake.freshness` | Duration | 3min | Maximum lag between Fluss and data lake table | -| `table.datalake.format` | String | - | Data lake format (paimon, iceberg, lance). Inherits from cluster config | +| `table.datalake.format` | String | - | Data lake format (paimon, iceberg, hudi, lance). Inherits from cluster config | | `table.datalake.auto-compaction` | Boolean | false | Auto-trigger compaction in the data lake | | `table.datalake.auto-expire-snapshot` | Boolean | false | Auto-expire snapshots in the data lake | diff --git a/website/docs/maintenance/tiered-storage/overview.mdx b/website/docs/maintenance/tiered-storage/overview.mdx index 3778baa915..bd2d13123b 100644 --- a/website/docs/maintenance/tiered-storage/overview.mdx +++ b/website/docs/maintenance/tiered-storage/overview.mdx @@ -15,7 +15,7 @@ Fluss organizes data into different storage layers based on its access patterns, Fluss ensures the recent data is stored in local for higher write/read performance and the historical data is stored in [remote storage](remote-storage.md) for lower cost. What's more, since the native format of Fluss's data is optimized for real-time write/read which is inevitable unfriendly to batch analytics, Fluss also introduces a [lakehouse storage](lakehouse-storage.md) which stores the data -in the well-known open data lake format for better analytics performance. Currently, supported formats are Paimon, Iceberg, and Lance. In the future, more kinds of data lake support are on the way. Keep eyes on us! +in the well-known open data lake format for better analytics performance. Currently, supported formats are Paimon, Iceberg, Hudi, and Lance. In the future, more kinds of data lake support are on the way. Keep eyes on us! The overall tiered storage architecture is shown in the following diagram: diff --git a/website/docs/streaming-lakehouse/datalake-formats/hudi.md b/website/docs/streaming-lakehouse/datalake-formats/hudi.md new file mode 100644 index 0000000000..6804e1b15c --- /dev/null +++ b/website/docs/streaming-lakehouse/datalake-formats/hudi.md @@ -0,0 +1,410 @@ +--- +title: Hudi +sidebar_position: 4 +--- + +# Hudi + +## Introduction + +[Apache Hudi](https://hudi.apache.org/) is an open lakehouse table format that provides transactional writes, record-level updates, and incremental processing on data lakes. +To integrate Fluss with Hudi, you must enable lakehouse storage and configure Hudi as the lakehouse storage. For more details, see [Deploying Streaming Lakehouse](../../install-deploy/deploying-streaming-lakehouse.md). + +Fluss tiers data to standard Hudi tables. Primary-key Fluss tables are written as Hudi Merge-On-Read tables, while Fluss log tables are written as Hudi Copy-On-Write tables. + +## Dependencies + +Apache Fluss publishes the Hudi lake connector to Maven Central: + +| Artifact | Jar | +|----------|-----| +| Fluss Hudi lake connector | [fluss-lake-hudi-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-lake-hudi/$FLUSS_VERSION$/fluss-lake-hudi-$FLUSS_VERSION$.jar) | + +Maven coordinates: + +```xml + + org.apache.fluss + fluss-lake-hudi + $FLUSS_VERSION$ + +``` + +Verify downloaded JARs against the [KEYS file](https://downloads.apache.org/incubator/fluss/KEYS) using the [verification instructions](/downloads#verifying-downloads). + +## Version Compatibility + +| Component | Required/Tested Versions | +|-----------|--------------------------| +| Hudi | 1.1.0 | +| Flink tiering runtime | Flink 1.20 with `hudi-flink1.20-bundle-1.1.0.jar` | + +If you run the tiering service on another Flink version, use the Hudi Flink bundle that matches your Flink runtime. + +## Configure Hudi as LakeHouse Storage + +### Configure Hudi in Cluster Configurations + +To configure Hudi as the lakehouse storage, configure the following options in `server.yaml`. +The example below uses Hudi's DFS catalog: + +```yaml +# Hudi configuration +datalake.enabled: true +datalake.format: hudi + +# Hudi catalog configuration, using DFS catalog mode +datalake.hudi.mode: dfs +datalake.hudi.catalog.path: /tmp/hudi +``` + +The directory configured by `datalake.hudi.catalog.path` must exist and be accessible to the Fluss servers. + +Fluss processes Hudi configurations by stripping the `datalake.hudi.` prefix and passing the remaining options to Hudi. +For example, `datalake.hudi.catalog.path` is passed to Hudi as `catalog.path`. +Cluster-level Hudi catalog options use the `datalake.hudi.*` prefix, while table-level Hudi options use the `hudi.*` prefix. +Fluss strips the corresponding prefix before passing these options to Hudi. + +Fluss supports the Hudi catalog modes implemented by the connector: + +- `dfs`: Hudi DFS catalog. Tables are stored under `${catalog.path}/${database_name}/${table_name}`. +- `hms`: Hudi Hive Metastore catalog. Configure Hive Metastore access through Hudi/Hadoop options, for example `hive.conf.dir` or `hadoop.hive.metastore.uris`. + +Example using Hive Metastore catalog mode: + +```yaml +datalake.enabled: true +datalake.format: hudi +datalake.hudi.mode: hms +datalake.hudi.catalog.path: hdfs:///warehouse/hudi +datalake.hudi.hadoop.hive.metastore.uris: thrift://:9083 +``` + +### Prepare Server-Side JARs + +Put the following JARs into `${FLUSS_HOME}/plugins/hudi/` before starting the Fluss servers: + +- [fluss-lake-hudi-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-lake-hudi/$FLUSS_VERSION$/fluss-lake-hudi-$FLUSS_VERSION$.jar) +- The Hudi Flink bundle matching the Flink version used by the Hudi integration. For Flink 1.20, use [hudi-flink1.20-bundle-1.1.0.jar](https://repo.maven.apache.org/maven2/org/apache/hudi/hudi-flink1.20-bundle/1.1.0/hudi-flink1.20-bundle-1.1.0.jar). +- Flink core and table runtime JARs matching the Hudi Flink bundle, including `flink-core`, `flink-table-common`, `flink-table-api-java`, and `flink-table-runtime`. +- Any Hadoop, Hive, or filesystem dependencies required by your Hudi catalog and table storage. + +Restart Fluss after changing the plugin directory. + +### Start Tiering Service to Hudi + +Then, start the datalake tiering service to tier Fluss data to Hudi. For the general process, see [Deploying Streaming Lakehouse](../../install-deploy/deploying-streaming-lakehouse.md). + +For Hudi, prepare the following JARs in `${FLINK_HOME}/lib`: + +- Put the Fluss Flink connector JAR into `${FLINK_HOME}/lib`; pick the connector matching your Flink version (see [Dependencies](../../engine-flink/getting-started.md#dependencies)). For Flink 1.20, use [fluss-flink-1.20-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-1.20/$FLUSS_VERSION$/fluss-flink-1.20-$FLUSS_VERSION$.jar). +- If you are using [Amazon S3](http://aws.amazon.com/s3/), [Aliyun OSS](https://www.aliyun.com/product/oss), or [HDFS](https://hadoop.apache.org/docs/stable/) as Fluss's [remote storage](../../maintenance/tiered-storage/remote-storage.md), download the corresponding Fluss filesystem JAR (see the [Filesystems](../../maintenance/tiered-storage/filesystems/overview.md) section) and put it into `${FLINK_HOME}/lib`. +- Put [fluss-lake-hudi-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-lake-hudi/$FLUSS_VERSION$/fluss-lake-hudi-$FLUSS_VERSION$.jar) into `${FLINK_HOME}/lib`. +- Put the Hudi Flink bundle into `${FLINK_HOME}/lib`. For Flink 1.20, use [hudi-flink1.20-bundle-1.1.0.jar](https://repo.maven.apache.org/maven2/org/apache/hudi/hudi-flink1.20-bundle/1.1.0/hudi-flink1.20-bundle-1.1.0.jar). +- Put any Hadoop, Hive, or filesystem dependencies required by your Hudi catalog and table storage into `${FLINK_HOME}/lib`. + +Start the Flink tiering job with Hudi-specific configurations: + +```shell +/bin/flink run /path/to/fluss-flink-tiering-$FLUSS_VERSION$.jar \ + --fluss.bootstrap.servers localhost:9123 \ + --datalake.format hudi \ + --datalake.hudi.mode dfs \ + --datalake.hudi.catalog.path /tmp/hudi +``` + +For Hive Metastore catalog mode: + +```shell +/bin/flink run /path/to/fluss-flink-tiering-$FLUSS_VERSION$.jar \ + --fluss.bootstrap.servers localhost:9123 \ + --datalake.format hudi \ + --datalake.hudi.mode hms \ + --datalake.hudi.catalog.path hdfs:///warehouse/hudi \ + --datalake.hudi.hadoop.hive.metastore.uris thrift://:9083 +``` + +Then, the datalake tiering service continuously tiers data from Fluss to Hudi. The table option `table.datalake.freshness` controls the target freshness of Hudi tables. By default, the data freshness is 3 minutes. + +## Table Mapping Between Fluss and Hudi + +When a Fluss table is created with the option `'table.datalake.enabled' = 'true'` and configured with Hudi as the datalake format, Fluss automatically creates a corresponding Hudi table with the same database and table name. + +For DFS catalog mode, the Hudi table path is `${catalog.path}/${database_name}/${table_name}` unless the Hudi table path is explicitly set by Hudi options. +For Hive Metastore catalog mode, the table path follows Hudi Hive catalog path inference. + +The schema of the Hudi table matches the Fluss table schema, except for three system columns appended by Fluss: + +| Column | Type | Description | +|---------------|--------------|-----------------------------------------------| +| `__bucket` | INT | Fluss bucket identifier for data distribution | +| `__offset` | BIGINT | Fluss log offset for ordering and seeking | +| `__timestamp` | TIMESTAMP(6) | Fluss log timestamp | + +Do not use user columns named `__bucket`, `__offset`, or `__timestamp`. Hudi metadata column names starting with `_hoodie_` are also reserved. + +### Primary Key Tables + +Primary-key Fluss tables are mapped to Hudi Merge-On-Read tables: + +- Hudi table type is set to `MERGE_ON_READ`. +- Hudi record key fields are derived from the Fluss primary key. +- Hudi bucket index is enabled and uses the Fluss bucket number. +- If Fluss bucket keys are configured, Hudi bucket index key fields are derived from the Fluss bucket keys. Otherwise Hudi uses the record key fields for bucket indexing. +- Writes use Hudi upsert semantics. + +Example: + +```sql title="Flink SQL" +USE CATALOG fluss_catalog; + +CREATE TABLE user_profiles ( + `user_id` BIGINT, + `username` STRING, + `email` STRING, + `last_login` TIMESTAMP, + `profile_data` STRING, + PRIMARY KEY (`user_id`) NOT ENFORCED +) WITH ( + 'table.datalake.enabled' = 'true', + 'table.datalake.freshness' = '30s', + 'bucket.num' = '4', + 'bucket.key' = 'user_id', + 'hudi.precombine.field' = 'last_login' +); +``` + +`hudi.precombine.field` is a Hudi table option used by Hudi to order records with the same record key during upsert. Choose a field that reflects your update ordering. + +Conceptually, Fluss creates a Hudi table with properties equivalent to: + +```sql title="Hudi table properties" +'connector' = 'hudi', +'table.type' = 'MERGE_ON_READ', +'hoodie.datasource.write.recordkey.field' = 'user_id', +'index.type' = 'BUCKET', +'hoodie.bucket.index.hash.field' = 'user_id', +'hoodie.bucket.index.num.buckets' = '4', +'precombine.field' = 'last_login' +``` + +### Log Tables + +Fluss log tables are mapped to Hudi Copy-On-Write tables: + +- Hudi table type is set to `COPY_ON_WRITE`. +- Hudi bucket index is enabled and uses the Fluss bucket number. +- Writes use Hudi insert semantics. +- A Hudi record key field must be configured explicitly because Fluss log tables do not have a primary key. + +Use the `hudi.` prefix to set Hudi table options. For log tables, configure `hudi.hoodie.datasource.write.recordkey.field`. +If you also configure Fluss bucket keys, the Hudi bucket index key must be a subset of the Hudi record key fields. + +Example: + +```sql title="Flink SQL" +CREATE TABLE access_logs ( + `event_id` STRING, + `user_id` BIGINT, + `action` STRING, + `event_time` TIMESTAMP +) WITH ( + 'table.datalake.enabled' = 'true', + 'table.datalake.freshness' = '30s', + 'bucket.num' = '8', + 'bucket.key' = 'event_id', + 'hudi.hoodie.datasource.write.recordkey.field' = 'event_id', + 'hudi.precombine.field' = 'event_time' +); +``` + +Conceptually, Fluss creates a Hudi table with properties equivalent to: + +```sql title="Hudi table properties" +'connector' = 'hudi', +'table.type' = 'COPY_ON_WRITE', +'hoodie.datasource.write.recordkey.field' = 'event_id', +'index.type' = 'BUCKET', +'hoodie.bucket.index.hash.field' = 'event_id', +'hoodie.bucket.index.num.buckets' = '8', +'precombine.field' = 'event_time' +``` + +### Partitioned Tables + +For Fluss partitioned tables, Fluss uses the Fluss partition keys as Hudi partition path fields. +The Hudi partition path field is managed by Fluss and should not be set manually. + +```sql title="Flink SQL" +CREATE TABLE daily_orders ( + `order_id` BIGINT, + `amount` DECIMAL(10, 2), + `order_date` STRING, + `updated_at` TIMESTAMP, + PRIMARY KEY (`order_id`, `order_date`) NOT ENFORCED +) PARTITIONED BY (`order_date`) +WITH ( + 'table.datalake.enabled' = 'true', + 'table.datalake.freshness' = '30s', + 'bucket.num' = '4', + 'bucket.key' = 'order_id', + 'hudi.precombine.field' = 'updated_at' +); +``` + +### Hudi Table Properties + +You can specify Hudi table properties when creating a datalake-enabled Fluss table by using the `hudi.` prefix within the Fluss table properties clause. +Fluss strips the `hudi.` prefix before passing the option to Hudi. + +```sql title="Flink SQL" +CREATE TABLE orders_with_lake ( + `order_id` BIGINT, + `customer_id` BIGINT, + `amount` DECIMAL(10, 2), + `event_time` TIMESTAMP, + PRIMARY KEY (`order_id`) NOT ENFORCED +) WITH ( + 'table.datalake.enabled' = 'true', + 'table.datalake.freshness' = '30s', + 'hudi.precombine.field' = 'event_time', + 'hudi.write.batch.size' = '64', + 'hudi.compaction.delta_commits' = '1' +); +``` + +Fluss manages the following Hudi behaviors automatically: table type, record key fields for primary-key tables, index type, bucket index key fields, bucket count, and partition path fields. Do not override these options manually. The only exception is that log tables must provide `hoodie.datasource.write.recordkey.field`. + +## Read Tables + +### Union Read with Apache Flink + +When a table has `'table.datalake.enabled' = 'true'`, its data exists in two layers: + +- Fresh data is retained in Fluss. +- Historical data is tiered to Hudi. + +To read the full dataset, query the Fluss table without any suffix. Fluss performs union read across the Hudi layer and the Fluss layer. + +```sql title="Flink SQL" +-- Set execution mode to streaming or batch; batch mode is used here as an example. +SET 'execution.runtime-mode' = 'batch'; + +-- Query will union data from Fluss and Hudi. +SELECT COUNT(*) FROM access_logs; +``` + +Union read supports both batch and streaming modes for Hudi log tables and primary-key tables. +In streaming mode, Flink first reads the latest readable Hudi instant tiered by the tiering service, then switches to Fluss from the corresponding log offsets. + +Key behavior for data retention: + +- Expired Fluss log data, controlled by `table.log.ttl`, remains accessible through Hudi if it was tiered before expiration. +- Cleaned-up partitions in partitioned tables, controlled by `table.auto-partition.num-retention`, remain accessible through Hudi if they were tiered before cleanup. + +### Read Hudi Tables Directly + +The `$lake` table suffix in the Fluss catalog currently supports Paimon and Iceberg only. To read Hudi-only data, use a native Hudi catalog or any engine that supports Hudi tables. + +Example using Hudi's Flink catalog with DFS catalog mode: + +```sql title="Flink SQL" +CREATE CATALOG hudi_catalog WITH ( + 'type' = 'hudi', + 'mode' = 'dfs', + 'catalog.path' = '/tmp/hudi' +); + +USE CATALOG hudi_catalog; + +SELECT COUNT(*) FROM fluss.access_logs; +``` + +For Hive Metastore catalog mode: + +```sql title="Flink SQL" +CREATE CATALOG hudi_catalog WITH ( + 'type' = 'hudi', + 'mode' = 'hms', + 'catalog.path' = 'hdfs:///warehouse/hudi', + 'hadoop.hive.metastore.uris' = 'thrift://:9083' +); +``` + +The Hudi catalog options must match the `datalake.hudi.*` options used by Fluss and the tiering service. + +## Data Type Mapping + +When integrating with Hudi, Fluss converts Fluss data types to Flink data types accepted by Hudi's Flink catalog: + +| Fluss Data Type | Hudi/Flink Data Type | Notes | +|-------------------------------|---------------------------------------|-------| +| BOOLEAN | BOOLEAN | | +| TINYINT | TINYINT | | +| SMALLINT | SMALLINT | | +| INT | INT | | +| BIGINT | BIGINT | | +| FLOAT | FLOAT | | +| DOUBLE | DOUBLE | | +| DECIMAL | DECIMAL | | +| STRING | STRING | | +| CHAR | STRING | Converted to STRING | +| DATE | DATE | | +| TIME | TIME | | +| TIMESTAMP | TIMESTAMP | | +| TIMESTAMP WITH LOCAL TIMEZONE | TIMESTAMP WITH LOCAL TIME ZONE / BIGINT | Uses TIMESTAMP WITH LOCAL TIME ZONE in DFS catalog mode and BIGINT in HMS catalog mode | +| BINARY | BINARY | | +| BYTES | BYTES | | +| ARRAY | ARRAY | | +| MAP | MAP | | +| ROW | ROW | | + +## Maintenance and Optimization + +### Auto Compaction + +The table option `table.datalake.auto-compaction` enables automatic Hudi compaction for primary-key tables. +It is disabled by default. + +When auto compaction is enabled for a Hudi Merge-On-Read table, the tiering service schedules, executes, and commits Hudi compaction during tiering. +Compaction does not apply to Hudi Copy-On-Write log tables. + +```sql title="Flink SQL" +CREATE TABLE user_profiles ( + `user_id` BIGINT, + `username` STRING, + `updated_at` TIMESTAMP, + PRIMARY KEY (`user_id`) NOT ENFORCED +) WITH ( + 'table.datalake.enabled' = 'true', + 'table.datalake.auto-compaction' = 'true', + 'hudi.precombine.field' = 'updated_at', + 'hudi.compaction.delta_commits' = '1' +); +``` + +The tiering service also accepts Hudi compaction timeout options through Hudi table properties: + +| Option | Default | Description | +|--------|---------|-------------| +| `hudi.fluss.tiering.compaction.complete-timeout` | `30min` | Maximum time for a tiering writer to wait for Hudi compaction execution to finish. | +| `hudi.fluss.tiering.compaction.shutdown-timeout` | `30s` | Maximum time to wait when shutting down the Hudi compaction executor. | + +### Commit Metadata + +Fluss adds metadata to Hudi commits for traceability and recovery: + +- `commit-user`: Set to `__fluss_lake_tiering` to identify Hudi instants committed by Fluss. +- `fluss-offsets`: A Fluss-managed offset metadata file path. The file records the Fluss bucket log-end offsets covered by the Hudi instant. + +The Fluss lake snapshot ID corresponds to the committed Hudi instant time. During recovery, Fluss finds the latest completed Hudi instant whose `commit-user` is `__fluss_lake_tiering` and uses its extra metadata to recover any lake snapshot that was committed to Hudi but not yet committed back to Fluss. + +## Current Limitations + +- Hudi lake catalog supports `dfs` and `hms` modes. +- Hudi table alteration through Fluss lake catalog is not supported yet. +- The Fluss catalog `$lake` suffix does not expose Hudi-only tables yet. Use Hudi's native catalog to read Hudi-only data. +- Fluss log tables must configure `hudi.hoodie.datasource.write.recordkey.field`. +- Hudi bucket key fields must be scalar types with deterministic string representations. Composite and binary types such as ARRAY, MAP, ROW, BINARY, and BYTES are not supported as Hudi bucket keys. +- For composite Hudi bucket keys, values containing `,` or colliding with Hudi's reserved placeholders `__null__` and `__empty__` are rejected to keep Fluss bucket routing aligned with Hudi bucket IDs. +- Sorted lake reads for primary-key union read are supported only for Hudi Merge-On-Read tables and require the query projection to include all Hudi record key fields. diff --git a/website/docs/streaming-lakehouse/overview.mdx b/website/docs/streaming-lakehouse/overview.mdx index 8ec28b0728..e9281a72d7 100644 --- a/website/docs/streaming-lakehouse/overview.mdx +++ b/website/docs/streaming-lakehouse/overview.mdx @@ -48,7 +48,7 @@ Some powerful features it provides are: - **Analytical Streams**: The union reads help data streams to have the powerful analytics capabilities. This reduces complexity when developing streaming applications, simplifies debugging, and allows for immediate access to live data insights. - **Connect to Lakehouse Ecosystem**: Fluss keeps the table metadata in sync with data lake catalogs while compacting data into Lakehouse. As a result, external engines like Spark, StarRocks, Flink, and Trino can read the data directly. They simply connect to the data lake catalog. -Currently, Fluss supports [Paimon](datalake-formats/paimon.md), [Iceberg](datalake-formats/iceberg.md), and [Lance](datalake-formats/lance.md) as Lakehouse Storage. Support for additional data lake formats is on the roadmap. +Currently, Fluss supports [Paimon](datalake-formats/paimon.md), [Iceberg](datalake-formats/iceberg.md), [Hudi](datalake-formats/hudi.md), and [Lance](datalake-formats/lance.md) as Lakehouse Storage. Support for additional data lake formats is on the roadmap. ## Getting Started diff --git a/website/docs/streaming-lakehouse/tiering-service.md b/website/docs/streaming-lakehouse/tiering-service.md index f92815931d..8e4e6c0629 100644 --- a/website/docs/streaming-lakehouse/tiering-service.md +++ b/website/docs/streaming-lakehouse/tiering-service.md @@ -5,13 +5,13 @@ sidebar_position: 2 # Tiering Service -The Tiering Service continuously compacts real-time data from Fluss into the configured lake format (Paimon, Iceberg, or Lance) for cost-efficient long-term storage and analytics. +The Tiering Service continuously compacts real-time data from Fluss into the configured lake format (Paimon, Iceberg, Hudi, or Lance) for cost-efficient long-term storage and analytics. ## Overview The Tiering Service is implemented as an Apache Flink job that: - Reads records from Fluss tables with lakehouse storage enabled -- Writes data to the configured data lake format (Paimon, Iceberg, or Lance) +- Writes data to the configured data lake format (Paimon, Iceberg, Hudi, or Lance) - Maintains exactly-once semantics between Fluss and the data lake - Operates incrementally, syncing only missing data segments diff --git a/website/docs/streaming-lakehouse/union-read.md b/website/docs/streaming-lakehouse/union-read.md index e0e901f8fa..b78f27376d 100644 --- a/website/docs/streaming-lakehouse/union-read.md +++ b/website/docs/streaming-lakehouse/union-read.md @@ -31,7 +31,7 @@ SELECT COUNT(*), SUM(amount) FROM orders; ### Lake-Only Read -To query only the data stored in the data lake, use the `$lake` suffix: +For formats exposed through the Fluss lake catalog, such as Paimon and Iceberg, use the `$lake` suffix to query only the data stored in the data lake: ```sql title="Flink SQL" -- Lake-only read: queries only tiered data @@ -41,6 +41,10 @@ SELECT * FROM my_table$lake; SELECT snapshot_id, total_record_count FROM my_table$lake$snapshots; ``` +:::note +The Fluss catalog does not expose Hudi-only tables through the `$lake` suffix yet. Use Hudi's native catalog or another Hudi-compatible engine for Hudi-only reads. +::: + Lake-only queries are useful when: - Real-time freshness is not required - You need to access lake format-specific system tables @@ -101,7 +105,7 @@ Key behavior for data retention with Union Read: | Engine | Union Read | Lake-Only Read | |--------|------------|----------------| -| Apache Flink | ✅ | ✅ Via `$lake` suffix | +| Apache Flink | ✅ | ✅ Via `$lake` suffix for Paimon/Iceberg; use native catalog for Hudi-only reads | | Apache Spark | ✅ | ✅ Via native lake connectors | | Trino | ❌ | ✅ Via native lake connectors | | StarRocks | ❌ | ✅ Via native lake connectors | @@ -111,3 +115,4 @@ For Spark union read usage, see [Spark - Reads](../engine-spark/reads.md#lake-en External engines can access the tiered lake data directly through native lake format connectors. See the specific data lake format documentation for examples: - [Paimon - Reading with other Engines](datalake-formats/paimon.md#reading-with-other-engines) - [Iceberg - Reading with other Engines](datalake-formats/iceberg.md#reading-with-other-engines) +- [Hudi - Read Hudi Tables Directly](datalake-formats/hudi.md#read-hudi-tables-directly)