apache / apache/hudi

Add a configuration to support flink reading and writing hudi tables across hadoop clusters

Open
#19,191 3 comments 0 reactions 0 assignees View on GitHub
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:**
Add a new Flink option parameter hadoop.conf.dir to load Hadoop configuration files, enabling Flink applications to read and write Hudi tables stored on HDFS across different Hadoop clusters, while isolating Flink’s HDFS rocksdb state storage from the Hadoop cluster configuration.

**Why this feature is needed:**
Because the Hadoop cluster used for Flink RocksDB state backend may be different from the Hadoop cluster where Hudi tables are stored, and Flink jobs may need to read and write Hudi tables across multiple Hadoop clusters, a new Hudi data source parameter is required to specify the path of Hadoop configuration files.

### User Experience

**How users will use this feature:**
- Configuration changes needed
Users can configure the hadoop.conf.dir property in the Hudi table parameters of Flink read/write jobs, specifying the Hadoop configuration directory for either the source or target Hadoop cluster.

- API changes
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java
```java
public static org.apache.hadoop.conf.Configuration getHadoopConf(Configuration conf) {
org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf();

// If hadoop.conf.dir is explicitly configured, load from that directory.
String hadoopConfDir = conf.getString(FlinkOptions.HADOOP_CONF_DIR.key(), null);
if (hadoopConfDir != null) {
org.apache.hadoop.conf.Configuration dirConf = loadHadoopConfFromDir(hadoopConfDir);
if (dirConf != null) {
hadoopConf = dirConf;
}
}

Map hadoopOptions = FlinkOptions.getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
hadoopOptions.remove("conf.dir");
hadoopOptions.forEach(hadoopConf::set);
Map ioOptions = OptionsResolver.getIOOptions(conf);
ioOptions.forEach(hadoopConf::set);
return hadoopConf;
}
```
- Usage examples
```sql
CREATE TABLE hudi_remote_table (...)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://remote-cluster:8020/path/to/table',
'hadoop.conf.dir' = '/path/to/remote-cluster/hadoop/conf',
...
)
```

### Hudi RFC Requirements

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

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

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java and trace FlinkOptions.getHadoopConf, loadHadoopConfFromDir, and OptionsResolver.getIOOptions. Verify how hadoop.conf.dir should coexist with Flink’s state-storage configuration and how source and target jobs obtain their Hadoop settings. Done means Flink jobs can read and write remote Hudi tables using the configured Hadoop directory without changing the separate state-storage configuration.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.