apache / apache/hudi

Support schema on write for flink

Open
#14,252 2 comments 0 reactions 0 assignees Claimed by @cshuo View on GitHub
area:schema engine:flink priority:low type:feature
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

### Feature Description

**What the feature achieves:**

Support out-of-box schema on write for flink.

**Why this feature is needed:**

Currently, spark writers support oob schema on write: https://hudi.apache.org/docs/schema_evolution/#schema-evolution-in-action, which allows user writing records to hudi table with backwards-compatible schema evolution scenarios, such as adding a null field or promoting field type. But flink writer don't have full support for schema on write, we should supplement this feature for flink writer.

E.g., the following writing to COW table with schema promoting is not supported.
```java
// writer with schema1
DataStream dataStream = execEnv.fromData(
createRowData("id1", "Alice", 25, Timestamp.from(Instant.now()), "par1"),
createRowData("id2", "Lily", 21, Timestamp.from(Instant.now()), "par1"),
createRowData("id3", "Julia", 15, Timestamp.from(Instant.now()), "par1")
);
HoodiePipeline.Builder builder = HoodiePipeline.builder("test_sink")
.column("uuid string not null")
.column("name string")
.column("age int")
.column("`ts` timestamp(3)")
.column("`partition` string")
.pk("uuid")
.partition("partition")
.options(options);
builder.sink(dataStream, false);
execute(execEnv, false, "Api_Sink_Test");

// writer with schema2, promoting `age` from INT to DOUBLE
dataStream = execEnv.fromData(
createRowData("id1", "Alice", 25.1, Timestamp.from(Instant.now().plusMillis(1000)), "par1"),
createRowData("id2", "Lily", 21.2, Timestamp.from(Instant.now().plusMillis(1000)), "par1"),
createRowData("id3", "Julia", 15.3, Timestamp.from(Instant.now().plusMillis(100)), "par1")
);
builder = HoodiePipeline.builder("test_sink")
.column("uuid string not null")
.column("name string")
.column("age double")
.column("`ts` timestamp(3)")
.column("`partition` string")
.pk("uuid")
.partition("partition")
.options(options);
builder.sink(dataStream, false);
execute(execEnv, false, "Api_Sink_Test_1");

```

### User Experience

**How users will use this feature:**
- Configuration changes needed
- API changes
- Usage examples

### Hudi RFC Requirements

**RFC PR link:** (if applicable)

**Why RFC is/isn't needed:**
- Does this change public interfaces/APIs? (Yes/No)
- Does this change storage format? (Yes/No)
- Justification:

Contributor guide

No contributing guide indexed for this repository

Research direction

Review open pull request #14248 first, then compare the Flink writer behavior with the Spark schema-on-write example linked in the issue. Use the HoodiePipeline.Builder example and its execute calls as the entry point; done means the two shown schemas, including INT-to-DOUBLE promotion, work for the Flink COW table.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.