cshuo commented on code in PR #19997: URL: https://github.com/apache/hudi/pull/19997#discussion_r4067807033
########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | Review Comment: SQL supports reader options via `WITH (...)`, e.g. `FROM STREAM current_orders WITH ('hoodie.datasource.query.incremental.format' = 'cdc')`. Start offsets and full-scan fallback options use the same clause, so Python is not required. Catalog resolution to Hudi’s existing streaming source and option propagation will be validated in Phase 3. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
