Add a configuration to support flink reading and writing hudi tables across hadoop clusters
- 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