You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ni...@apache.org on 2023/04/04 07:57:34 UTC
[incubator-paimon] branch master updated: [hadoop] Adjust Flink FileIO load (#826)
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 7b2345c7f [hadoop] Adjust Flink FileIO load (#826)
7b2345c7f is described below
commit 7b2345c7fd95bbe8f476a593fa71c7d9dd93a153
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Apr 4 15:57:28 2023 +0800
[hadoop] Adjust Flink FileIO load (#826)
---
docs/content/engines/flink.md | 4 +
docs/content/filesystems/hdfs.md | 90 +++++++++++
docs/content/filesystems/oss.md | 15 +-
docs/content/filesystems/overview.md | 17 ++-
docs/content/filesystems/s3.md | 15 +-
docs/content/project/download.md | 2 +-
.../org/apache/paimon/catalog/CatalogContext.java | 29 ++--
.../src/main/java/org/apache/paimon/fs/FileIO.java | 32 +++-
.../java/org/apache/paimon/fs/FileIOUtils.java | 30 ++--
.../java/org/apache/paimon/utils/HadoopUtils.java | 39 +----
.../apache/paimon/fs/HadoopConfigLoadingTest.java | 38 -----
.../paimon/flink/AbstractFlinkTableFactory.java | 4 +-
.../apache/paimon/flink/FlinkCatalogFactory.java | 4 +-
.../java/org/apache/paimon/flink/HadoopUtils.java | 164 ---------------------
14 files changed, 176 insertions(+), 307 deletions(-)
diff --git a/docs/content/engines/flink.md b/docs/content/engines/flink.md
index 2b979216a..36036fee6 100644
--- a/docs/content/engines/flink.md
+++ b/docs/content/engines/flink.md
@@ -85,6 +85,10 @@ cp paimon-flink-*.jar <FLINK_HOME>/lib/
**Step 3: Copy Hadoop Bundled Jar**
+{{< hint info >}}
+If the machine is in a hadoop environment, please ensure the value of the environment variable `HADOOP_CLASSPATH`, you do not need to use the following pre-bundled Hadoop jar.
+{{< /hint >}}
+
[Download](https://flink.apache.org/downloads.html) Pre-bundled Hadoop jar and copy the jar file to the `lib` directory of your Flink home.
```bash
diff --git a/docs/content/filesystems/hdfs.md b/docs/content/filesystems/hdfs.md
new file mode 100644
index 000000000..7c4eddc39
--- /dev/null
+++ b/docs/content/filesystems/hdfs.md
@@ -0,0 +1,90 @@
+---
+title: "HDFS"
+weight: 2
+type: docs
+aliases:
+- /filesystems/hdfs.html
+---
+<!--
+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.
+-->
+
+# HDFS
+
+You don't need any additional dependencies to access HDFS because you have already taken care of the Hadoop dependencies.
+
+## HDFS Configuration
+
+For HDFS, the most important thing is to be able to read your HDFS configuration.
+
+{{< tabs "hdfs conf" >}}
+
+{{< tab "Flink/Trino/JavaAPI" >}}
+
+You may not have to do anything, if you are in a hadoop environment. Otherwise pick one of the following ways to
+configure your HDFS:
+
+1. Set environment variable `HADOOP_HOME` or `HADOOP_CONF_DIR`.
+2. Configure `'fs.hdfs.hadoopconf'` in the paimon catalog.
+
+The first approach is recommended.
+
+{{< /tab >}}
+
+{{< tab "Hive/Spark" >}}
+
+HDFS Configuration is available directly through the computation cluster, see cluster configuration of Hive and Spark for details.
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+## Hadoop-compatible file systems (HCFS)
+
+All Hadoop file systems are automatically available when the Hadoop libraries are on the classpath.
+
+This way, Paimon seamlessly supports all of Hadoop file systems implementing the `org.apache.hadoop.fs.FileSystem`
+interface, and all Hadoop-compatible file systems (HCFS).
+
+- HDFS
+- Alluxio (see configuration specifics below)
+- XtreemFS
+- …
+
+The Hadoop configuration has to have an entry for the required file system implementation in the `core-site.xml` file.
+
+For Alluxio support add the following entry into the core-site.xml file:
+
+```shell
+<property>
+ <name>fs.alluxio.impl</name>
+ <value>alluxio.hadoop.FileSystem</value>
+</property>
+```
+
+## HDFS HA
+
+Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [HA configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HDFSHighAvailabilityWithNFS.html).
+
+## HDFS ViewFS
+
+Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [ViewFs configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/ViewFs.html).
+
+## Kerberos
+
+Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [Kerberos configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/SecureMode.html).
diff --git a/docs/content/filesystems/oss.md b/docs/content/filesystems/oss.md
index 8338ee76d..eebdc0b19 100644
--- a/docs/content/filesystems/oss.md
+++ b/docs/content/filesystems/oss.md
@@ -1,6 +1,6 @@
---
title: "OSS"
-weight: 2
+weight: 3
type: docs
aliases:
- /filesystems/oss.html
@@ -38,19 +38,6 @@ Download [paimon-oss-{{< version >}}.jar](https://repository.apache.org/snapshot
{{< /unstable >}}
-You can also manually build bundled jar from the source code.
-
-To build from source code, [clone the git repository]({{< github_repo >}}).
-
-Build shaded jar with the following command.
-
-```bash
-mvn clean install -DskipTests
-```
-
-You can find the shaded jars under
-`./paimon-filesystems/paimon-oss/target/paimon-oss-{{< version >}}.jar`.
-
{{< tabs "oss" >}}
{{< tab "Flink" >}}
diff --git a/docs/content/filesystems/overview.md b/docs/content/filesystems/overview.md
index dfa146ee1..7b1ac1dee 100644
--- a/docs/content/filesystems/overview.md
+++ b/docs/content/filesystems/overview.md
@@ -42,4 +42,19 @@ FileSystem pluggable jars for user to query tables from Spark/Hive side.
| Aliyun OSS | oss:// | Y | |
| S3 | s3:// | Y | |
-[Download Link]({{< ref "project/download#filesystem-jars" >}})
\ No newline at end of file
+## Dependency
+
+We recommend you to download the jar directly: [Download Link]({{< ref "project/download#filesystem-jars" >}}).
+
+You can also manually build bundled jar from the source code.
+
+To build from source code, [clone the git repository]({{< github_repo >}}).
+
+Build shaded jar with the following command.
+
+```bash
+mvn clean install -DskipTests
+```
+
+You can find the shaded jars under
+`./paimon-filesystems/paimon-${fs}/target/paimon-${fs}-{{< version >}}.jar`.
diff --git a/docs/content/filesystems/s3.md b/docs/content/filesystems/s3.md
index 005c53bd1..f4bd8c735 100644
--- a/docs/content/filesystems/s3.md
+++ b/docs/content/filesystems/s3.md
@@ -1,6 +1,6 @@
---
title: "S3"
-weight: 3
+weight: 4
type: docs
aliases:
- /filesystems/s3.html
@@ -38,19 +38,6 @@ Download [paimon-s3-{{< version >}}.jar](https://repository.apache.org/snapshots
{{< /unstable >}}
-You can also manually build bundled jar from the source code.
-
-To build from source code, [clone the git repository]({{< github_repo >}}).
-
-Build shaded jar with the following command.
-
-```bash
-mvn clean install -DskipTests
-```
-
-You can find the shaded jars under
-`./paimon-filesystems/paimon-s3/target/paimon-s3-{{< version >}}.jar`.
-
{{< tabs "oss" >}}
{{< tab "Flink" >}}
diff --git a/docs/content/project/download.md b/docs/content/project/download.md
index fea839ef5..4425638ab 100644
--- a/docs/content/project/download.md
+++ b/docs/content/project/download.md
@@ -1,6 +1,6 @@
---
title: "Download"
-weight: 100
+weight: 2
type: docs
aliases:
- /project/download.html
diff --git a/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java b/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
index a876f32c9..8b1c450d6 100644
--- a/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
+++ b/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
@@ -21,16 +21,15 @@ package org.apache.paimon.catalog;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.fs.FileIOLoader;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.hadoop.HadoopFileIOLoader;
import org.apache.paimon.options.Options;
-import org.apache.paimon.utils.HadoopUtils;
import org.apache.hadoop.conf.Configuration;
import javax.annotation.Nullable;
-import static org.apache.paimon.options.CatalogOptions.FS_ALLOW_HADOOP_FALLBACK;
import static org.apache.paimon.options.CatalogOptions.WAREHOUSE;
+import static org.apache.paimon.utils.HadoopUtils.getHadoopConfiguration;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
/**
* Context of catalog.
@@ -45,14 +44,12 @@ public class CatalogContext {
@Nullable private final FileIOLoader fallbackIOLoader;
private CatalogContext(
- Options options, Configuration hadoopConf, @Nullable FileIOLoader fallbackIOLoader) {
- this.options = options;
- this.hadoopConf = hadoopConf;
- if (fallbackIOLoader == null && options.get(FS_ALLOW_HADOOP_FALLBACK)) {
- this.fallbackIOLoader = new HadoopFileIOLoader();
- } else {
- this.fallbackIOLoader = fallbackIOLoader;
- }
+ Options options,
+ @Nullable Configuration hadoopConf,
+ @Nullable FileIOLoader fallbackIOLoader) {
+ this.options = checkNotNull(options);
+ this.hadoopConf = hadoopConf == null ? getHadoopConfiguration(options) : hadoopConf;
+ this.fallbackIOLoader = fallbackIOLoader;
}
public static CatalogContext create(Path warehouse) {
@@ -62,15 +59,19 @@ public class CatalogContext {
}
public static CatalogContext create(Options options) {
- return create(options, HadoopUtils.getHadoopConfiguration(options));
+ return new CatalogContext(options, null, null);
}
public static CatalogContext create(Options options, Configuration hadoopConf) {
- return create(options, hadoopConf, null);
+ return new CatalogContext(options, hadoopConf, null);
+ }
+
+ public static CatalogContext create(Options options, FileIOLoader fallbackIOLoader) {
+ return new CatalogContext(options, null, fallbackIOLoader);
}
public static CatalogContext create(
- Options options, Configuration hadoopConf, @Nullable FileIOLoader fallbackIOLoader) {
+ Options options, Configuration hadoopConf, FileIOLoader fallbackIOLoader) {
return new CatalogContext(options, hadoopConf, fallbackIOLoader);
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
index 08b62d65b..b401c454a 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
@@ -20,6 +20,7 @@ package org.apache.paimon.fs;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.hadoop.HadoopFileIOLoader;
import org.apache.paimon.fs.local.LocalFileIO;
import org.slf4j.Logger;
@@ -39,6 +40,8 @@ import java.util.Map;
import java.util.ServiceLoader;
import java.util.UUID;
+import static org.apache.paimon.fs.FileIOUtils.checkAccess;
+
/**
* File IO to read and write file.
*
@@ -244,14 +247,31 @@ public interface FileIO extends Serializable {
Map<String, FileIOLoader> loaders = discoverLoaders();
FileIOLoader loader = loaders.get(uri.getScheme());
+
+ // load fallbackIO
+ FileIOLoader fallbackIO = config.fallbackIO();
+ if (loader == null) {
+ loader = checkAccess(fallbackIO, path);
+ }
+
+ // load hadoopIO
+ if (loader == null) {
+ loader = checkAccess(new HadoopFileIOLoader(), path);
+ }
+
if (loader == null) {
- loader = config.fallbackIO();
- if (loader == null) {
- throw new UnsupportedSchemeException(
- String.format(
- "Could not find a file io implementation for scheme '%s' in the classpath.",
- uri.getScheme()));
+ String fallbackMsg = "";
+ if (fallbackIO != null) {
+ fallbackMsg =
+ " "
+ + fallbackIO.getClass().getSimpleName()
+ + " also cannot access this path.";
}
+ throw new UnsupportedSchemeException(
+ String.format(
+ "Could not find a file io implementation for scheme '%s' in the classpath."
+ + "%s Hadoop FileSystem also cannot access this path '%s'.",
+ uri.getScheme(), fallbackMsg, path));
}
FileIO fileIO = loader.load(path);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkUtils.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIOUtils.java
similarity index 56%
rename from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkUtils.java
rename to paimon-common/src/main/java/org/apache/paimon/fs/FileIOUtils.java
index 479a5ca1a..3a7670501 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIOUtils.java
@@ -16,24 +16,24 @@
* limitations under the License.
*/
-package org.apache.paimon.flink;
+package org.apache.paimon.fs;
-import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.options.Options;
+import java.io.IOException;
-import org.apache.flink.configuration.ReadableConfig;
+/** Utils for {@link FileIO}. */
+public class FileIOUtils {
-import java.util.Map;
+ public static FileIOLoader checkAccess(FileIOLoader fileIO, Path path) {
+ try {
+ if (fileIO == null) {
+ return null;
+ }
-/** Utils for Flink. */
-public class FlinkUtils {
-
- public static CatalogContext createCatalogContext(
- Map<String, String> options, ReadableConfig flinkConf) {
- return CatalogContext.create(
- Options.fromMap(options),
- HadoopUtils.getHadoopConfiguration(
- TableConfigUtils.extractConfiguration(flinkConf)),
- new FlinkFileIOLoader());
+ // check access
+ fileIO.load(path).exists(path);
+ return fileIO;
+ } catch (IOException ignore) {
+ return null;
+ }
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java
index 5197fac33..bfeb288cc 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java
@@ -38,26 +38,8 @@ public class HadoopUtils {
public static final String HADOOP_HOME_ENV = "HADOOP_HOME";
public static final String HADOOP_CONF_ENV = "HADOOP_CONF_DIR";
- /**
- * Path to hdfs-default.xml file.
- *
- * @deprecated Use environment variable HADOOP_CONF_DIR instead.
- */
- @Deprecated public static final String HDFS_DEFAULT_CONFIG = "fs.hdfs.hdfsdefault";
-
- /**
- * Path to hdfs-site.xml file.
- *
- * @deprecated Use environment variable HADOOP_CONF_DIR instead.
- */
- @Deprecated public static final String HDFS_SITE_CONFIG = "fs.hdfs.hdfssite";
-
- /**
- * Path to Hadoop configuration.
- *
- * @deprecated Use environment variable HADOOP_CONF_DIR instead.
- */
- @Deprecated public static final String PATH_HADOOP_CONFIG = "fs.hdfs.hadoopconf";
+ /** Path to Hadoop configuration. */
+ public static final String PATH_HADOOP_CONFIG = "fs.hdfs.hadoopconf";
public static Configuration getHadoopConfiguration(Options options) {
@@ -90,23 +72,6 @@ public class HadoopUtils {
}
// Approach 2: Paimon configuration (deprecated)
- final String hdfsDefaultPath = options.getString(HDFS_DEFAULT_CONFIG, null);
- if (hdfsDefaultPath != null) {
- result.addResource(new org.apache.hadoop.fs.Path(hdfsDefaultPath));
- LOG.debug(
- "Using hdfs-default configuration-file path from Paimon config: {}",
- hdfsDefaultPath);
- foundHadoopConfiguration = true;
- }
-
- final String hdfsSitePath = options.getString(HDFS_SITE_CONFIG, null);
- if (hdfsSitePath != null) {
- result.addResource(new org.apache.hadoop.fs.Path(hdfsSitePath));
- LOG.debug(
- "Using hdfs-site configuration-file path from Paimon config: {}", hdfsSitePath);
- foundHadoopConfiguration = true;
- }
-
final String hadoopConfigPath = options.getString(PATH_HADOOP_CONFIG, null);
if (hadoopConfigPath != null) {
LOG.debug(
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopConfigLoadingTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopConfigLoadingTest.java
index 50816dc5a..2a998a246 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopConfigLoadingTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopConfigLoadingTest.java
@@ -57,34 +57,6 @@ public class HadoopConfigLoadingTest {
assertEquals(IN_CP_CONFIG_VALUE, hadoopConf.get(IN_CP_CONFIG_KEY, null));
}
- @Test
- public void loadFromLegacyConfigEntries() throws Exception {
- final String k1 = "k1 prop";
- final String v1 = "value1";
-
- final String k2 = "k2 prop";
- final String v2 = "value2";
-
- final File file1 = tempDir.resolve("core-site.xml").toFile();
- final File file2 = tempDir.resolve("hdfs-site.xml").toFile();
-
- printConfig(file1, k1, v1);
- printConfig(file2, k2, v2);
-
- final Options options = new Options();
- options.set(HadoopUtils.HDFS_DEFAULT_CONFIG, file1.getAbsolutePath());
- options.set(HadoopUtils.HDFS_SITE_CONFIG, file2.getAbsolutePath());
-
- Configuration hadoopConf = HadoopUtils.getHadoopConfiguration(options);
-
- // contains extra entries
- assertEquals(v1, hadoopConf.get(k1, null));
- assertEquals(v2, hadoopConf.get(k2, null));
-
- // also contains classpath defaults
- assertEquals(IN_CP_CONFIG_VALUE, hadoopConf.get(IN_CP_CONFIG_KEY, null));
- }
-
@Test
public void loadFromHadoopConfEntry() throws Exception {
final String k1 = "singing?";
@@ -189,7 +161,6 @@ public class HadoopConfigLoadingTest {
final String v1 = "from HADOOP_CONF_DIR";
final String v2 = "from Paimon config `fs.hdfs.hadoopconf`";
- final String v3 = "from Paimon config `fs.hdfs.hdfsdefault`";
final String v4 = "from HADOOP_HOME/etc/hadoop";
final String v5 = "from HADOOP_HOME/conf";
@@ -209,7 +180,6 @@ public class HadoopConfigLoadingTest {
final File file1 = new File(hadoopConfDir, "core-site.xml");
final File file2 = new File(hadoopConfEntryDir, "core-site.xml");
- final File file3 = new File(legacyConfDir, "core-site.xml");
final File file4 = new File(hadoopHomeEtc, "core-site.xml");
final File file5 = new File(hadoopHomeConf, "core-site.xml");
@@ -220,12 +190,6 @@ public class HadoopConfigLoadingTest {
properties2.put(k2, v2);
printConfigs(file2, properties2);
- Map<String, String> properties3 = new HashMap<>();
- properties3.put(k1, v3);
- properties3.put(k2, v3);
- properties3.put(k3, v3);
- printConfigs(file3, properties3);
-
Map<String, String> properties4 = new HashMap<>();
properties4.put(k1, v4);
properties4.put(k2, v4);
@@ -243,7 +207,6 @@ public class HadoopConfigLoadingTest {
final Options options = new Options();
options.setString(HadoopUtils.PATH_HADOOP_CONFIG, hadoopConfEntryDir.getAbsolutePath());
- options.setString(HadoopUtils.HDFS_DEFAULT_CONFIG, file3.getAbsolutePath());
final Configuration hadoopConf;
@@ -261,7 +224,6 @@ public class HadoopConfigLoadingTest {
// contains extra entries
assertEquals(v1, hadoopConf.get(k1, null));
assertEquals(v2, hadoopConf.get(k2, null));
- assertEquals(v3, hadoopConf.get(k3, null));
assertEquals(v4, hadoopConf.get(k4, null));
assertEquals(v5, hadoopConf.get(k5, null));
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
index b7ea563ae..046289eb9 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
@@ -135,8 +135,8 @@ public abstract class AbstractFlinkTableFactory
}
static CatalogContext createCatalogContext(DynamicTableFactory.Context context) {
- return FlinkUtils.createCatalogContext(
- context.getCatalogTable().getOptions(), context.getConfiguration());
+ return CatalogContext.create(
+ Options.fromMap(context.getCatalogTable().getOptions()), new FlinkFileIOLoader());
}
static Table buildPaimonTable(DynamicTableFactory.Context context) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
index 94aa3d91f..982cff2ae 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
@@ -23,6 +23,7 @@ import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
+import org.apache.paimon.options.Options;
import java.util.Collections;
import java.util.Set;
@@ -56,7 +57,8 @@ public class FlinkCatalogFactory implements org.apache.flink.table.factories.Cat
public FlinkCatalog createCatalog(Context context) {
return createCatalog(
context.getName(),
- FlinkUtils.createCatalogContext(context.getOptions(), context.getConfiguration()),
+ CatalogContext.create(
+ Options.fromMap(context.getOptions()), new FlinkFileIOLoader()),
context.getClassLoader());
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/HadoopUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/HadoopUtils.java
deleted file mode 100644
index 7543bb997..000000000
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/HadoopUtils.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.paimon.flink;
-
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-
-/**
- * Utility class for working with Hadoop-related classes. This should only be used if Hadoop is on
- * the classpath.
- */
-public class HadoopUtils {
-
- private static final Logger LOG = LoggerFactory.getLogger(HadoopUtils.class);
-
- /** The prefixes that Flink adds to the Hadoop config. */
- private static final String[] FLINK_CONFIG_PREFIXES = {"flink.hadoop."};
-
- @SuppressWarnings("deprecation")
- public static Configuration getHadoopConfiguration(
- org.apache.flink.configuration.Configuration flinkConfiguration) {
-
- // Instantiate an HdfsConfiguration to load the hdfs-site.xml and hdfs-default.xml
- // from the classpath
-
- Configuration result = new HdfsConfiguration();
- boolean foundHadoopConfiguration = false;
-
- // We need to load both core-site.xml and hdfs-site.xml to determine the default fs path and
- // the hdfs configuration.
- // The properties of a newly added resource will override the ones in previous resources, so
- // a configuration
- // file with higher priority should be added later.
-
- // Approach 1: HADOOP_HOME environment variables
- String[] possibleHadoopConfPaths = new String[2];
-
- final String hadoopHome = System.getenv("HADOOP_HOME");
- if (hadoopHome != null) {
- LOG.debug("Searching Hadoop configuration files in HADOOP_HOME: {}", hadoopHome);
- possibleHadoopConfPaths[0] = hadoopHome + "/conf";
- possibleHadoopConfPaths[1] = hadoopHome + "/etc/hadoop"; // hadoop 2.2
- }
-
- for (String possibleHadoopConfPath : possibleHadoopConfPaths) {
- if (possibleHadoopConfPath != null) {
- foundHadoopConfiguration = addHadoopConfIfFound(result, possibleHadoopConfPath);
- }
- }
-
- // Approach 2: Flink configuration (deprecated)
- final String hdfsDefaultPath =
- flinkConfiguration.getString(ConfigConstants.HDFS_DEFAULT_CONFIG, null);
- if (hdfsDefaultPath != null) {
- result.addResource(new org.apache.hadoop.fs.Path(hdfsDefaultPath));
- LOG.debug(
- "Using hdfs-default configuration-file path from Flink config: {}",
- hdfsDefaultPath);
- foundHadoopConfiguration = true;
- }
-
- final String hdfsSitePath =
- flinkConfiguration.getString(ConfigConstants.HDFS_SITE_CONFIG, null);
- if (hdfsSitePath != null) {
- result.addResource(new org.apache.hadoop.fs.Path(hdfsSitePath));
- LOG.debug(
- "Using hdfs-site configuration-file path from Flink config: {}", hdfsSitePath);
- foundHadoopConfiguration = true;
- }
-
- final String hadoopConfigPath =
- flinkConfiguration.getString(ConfigConstants.PATH_HADOOP_CONFIG, null);
- if (hadoopConfigPath != null) {
- LOG.debug("Searching Hadoop configuration files in Flink config: {}", hadoopConfigPath);
- foundHadoopConfiguration =
- addHadoopConfIfFound(result, hadoopConfigPath) || foundHadoopConfiguration;
- }
-
- // Approach 3: HADOOP_CONF_DIR environment variable
- String hadoopConfDir = System.getenv("HADOOP_CONF_DIR");
- if (hadoopConfDir != null) {
- LOG.debug("Searching Hadoop configuration files in HADOOP_CONF_DIR: {}", hadoopConfDir);
- foundHadoopConfiguration =
- addHadoopConfIfFound(result, hadoopConfDir) || foundHadoopConfiguration;
- }
-
- // Approach 4: Flink configuration
- // add all configuration key with prefix 'flink.hadoop.' in flink conf to hadoop conf
- for (String key : flinkConfiguration.keySet()) {
- for (String prefix : FLINK_CONFIG_PREFIXES) {
- if (key.startsWith(prefix)) {
- String newKey = key.substring(prefix.length());
- String value = flinkConfiguration.getString(key, null);
- result.set(newKey, value);
- LOG.debug(
- "Adding Flink config entry for {} as {}={} to Hadoop config",
- key,
- newKey,
- value);
- foundHadoopConfiguration = true;
- }
- }
- }
-
- if (!foundHadoopConfiguration) {
- LOG.warn(
- "Could not find Hadoop configuration via any of the supported methods "
- + "(Flink configuration, environment variables).");
- }
-
- return result;
- }
-
- /**
- * Search Hadoop configuration files in the given path, and add them to the configuration if
- * found.
- */
- private static boolean addHadoopConfIfFound(
- Configuration configuration, String possibleHadoopConfPath) {
- boolean foundHadoopConfiguration = false;
- if (new File(possibleHadoopConfPath).exists()) {
- if (new File(possibleHadoopConfPath + "/core-site.xml").exists()) {
- configuration.addResource(
- new org.apache.hadoop.fs.Path(possibleHadoopConfPath + "/core-site.xml"));
- LOG.debug(
- "Adding "
- + possibleHadoopConfPath
- + "/core-site.xml to hadoop configuration");
- foundHadoopConfiguration = true;
- }
- if (new File(possibleHadoopConfPath + "/hdfs-site.xml").exists()) {
- configuration.addResource(
- new org.apache.hadoop.fs.Path(possibleHadoopConfPath + "/hdfs-site.xml"));
- LOG.debug(
- "Adding "
- + possibleHadoopConfPath
- + "/hdfs-site.xml to hadoop configuration");
- foundHadoopConfiguration = true;
- }
- }
- return foundHadoopConfiguration;
- }
-}