-
Notifications
You must be signed in to change notification settings - Fork 304
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[#5019] feat: (hadoop-catalog): Add a framework to support multi-storage in a pluginized manner for fileset catalog #5020
base: main
Are you sure you want to change the base?
Changes from 9 commits
d2447a2
7e5a8b5
f53c5ef
36fedcd
e93fba5
b1e04b6
db00e65
c793582
013f5cb
dba5753
16dfc73
278fcd8
3fb55ad
cd04666
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,53 @@ | ||
/* | ||
* 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.gravitino.catalog.hadoop; | ||
|
||
import java.io.IOException; | ||
import org.apache.hadoop.conf.Configuration; | ||
import org.apache.hadoop.fs.FileSystem; | ||
import org.apache.hadoop.fs.Path; | ||
|
||
public interface FileSystemProvider { | ||
|
||
/** | ||
* Get the FileSystem instance according to the configuration and the path. | ||
* | ||
* <p>Compared to the FileSystem.get method, this method allows the provider to create a | ||
* FileSystem instance with a specific configuration and path and do further initialization if | ||
* needed. | ||
* | ||
* <p>For example, we can check endpoint configurations for S3AFileSystem, or set the default one. | ||
* | ||
* @param configuration The configuration. | ||
* @param path The path. | ||
* @return The FileSystem instance. | ||
* @throws IOException If the FileSystem instance cannot be created. | ||
*/ | ||
FileSystem getFileSystem(Configuration configuration, Path path) throws IOException; | ||
|
||
/** | ||
* Get the scheme of this FileSystem provider. | ||
* | ||
* @return The scheme of this FileSystem provider. | ||
*/ | ||
default String getScheme() { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It doesn't make sense to return "file" as default, as there's no default filesystem implementation. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah, I removed the default implementation. |
||
return "file"; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -44,6 +44,8 @@ | |
import org.apache.gravitino.audit.CallerContext; | ||
import org.apache.gravitino.audit.FilesetAuditConstants; | ||
import org.apache.gravitino.audit.FilesetDataOperation; | ||
import org.apache.gravitino.catalog.hadoop.fs.HDFSFileSystemProvider; | ||
import org.apache.gravitino.catalog.hadoop.fs.LocalFileSystemProvider; | ||
import org.apache.gravitino.connector.CatalogInfo; | ||
import org.apache.gravitino.connector.CatalogOperations; | ||
import org.apache.gravitino.connector.HasPropertyMetadata; | ||
|
@@ -77,6 +79,7 @@ public class HadoopCatalogOperations implements CatalogOperations, SupportsSchem | |
private static final String SLASH = "/"; | ||
|
||
private static final Logger LOG = LoggerFactory.getLogger(HadoopCatalogOperations.class); | ||
public static final Map<String, FileSystemProvider> FILE_SYSTEM_PROVIDERS = Maps.newHashMap(); | ||
|
||
private final EntityStore store; | ||
|
||
|
@@ -90,6 +93,14 @@ public class HadoopCatalogOperations implements CatalogOperations, SupportsSchem | |
|
||
private CatalogInfo catalogInfo; | ||
|
||
static { | ||
FileSystemProvider localFileSystemProvider = new LocalFileSystemProvider(); | ||
FileSystemProvider hdfsFileSystemProvider = new HDFSFileSystemProvider(); | ||
|
||
FILE_SYSTEM_PROVIDERS.put(localFileSystemProvider.getScheme(), localFileSystemProvider); | ||
FILE_SYSTEM_PROVIDERS.put(hdfsFileSystemProvider.getScheme(), hdfsFileSystemProvider); | ||
} | ||
|
||
HadoopCatalogOperations(EntityStore store) { | ||
this.store = store; | ||
} | ||
|
@@ -119,32 +130,57 @@ public void initialize( | |
Map<String, String> config, CatalogInfo info, HasPropertyMetadata propertiesMetadata) | ||
throws RuntimeException { | ||
this.propertiesMetadata = propertiesMetadata; | ||
this.catalogInfo = info; | ||
|
||
// Initialize Hadoop Configuration. | ||
this.conf = config; | ||
this.hadoopConf = new Configuration(); | ||
this.catalogInfo = info; | ||
|
||
hadoopConf = new Configuration(); | ||
Map<String, String> bypassConfigs = | ||
config.entrySet().stream() | ||
conf.entrySet().stream() | ||
.filter(e -> e.getKey().startsWith(CATALOG_BYPASS_PREFIX)) | ||
.collect( | ||
Collectors.toMap( | ||
e -> e.getKey().substring(CATALOG_BYPASS_PREFIX.length()), | ||
Map.Entry::getValue)); | ||
bypassConfigs.forEach(hadoopConf::set); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The initialization of Hadoop conf here seems to be useless. It's better to move these codes above into the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah, you are right, I will do it according to your suggestion. |
||
|
||
initPluginFileSystem(conf); | ||
|
||
String catalogLocation = | ||
(String) | ||
propertiesMetadata | ||
.catalogPropertiesMetadata() | ||
.getOrDefault(config, HadoopCatalogPropertiesMetadata.LOCATION); | ||
conf.forEach(hadoopConf::set); | ||
|
||
this.catalogStorageLocation = | ||
StringUtils.isNotBlank(catalogLocation) | ||
? Optional.of(catalogLocation).map(Path::new) | ||
: Optional.empty(); | ||
} | ||
|
||
private void initPluginFileSystem(Map<String, String> config) { | ||
String fileSystemProviders = | ||
(String) | ||
propertiesMetadata | ||
.catalogPropertiesMetadata() | ||
.getOrDefault(config, HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDER); | ||
|
||
if (StringUtils.isNotBlank(fileSystemProviders)) { | ||
String[] providers = fileSystemProviders.split(","); | ||
for (String provider : providers) { | ||
try { | ||
FileSystemProvider fileSystemProvider = | ||
(FileSystemProvider) | ||
Class.forName(provider.trim()).getDeclaredConstructor().newInstance(); | ||
FILE_SYSTEM_PROVIDERS.put(fileSystemProvider.getScheme(), fileSystemProvider); | ||
} catch (Exception e) { | ||
throw new GravitinoRuntimeException( | ||
e, "Failed to initialize file system provider: %s", provider); | ||
} | ||
} | ||
} | ||
} | ||
|
||
@Override | ||
public NameIdentifier[] listFilesets(Namespace namespace) throws NoSuchSchemaException { | ||
try { | ||
|
@@ -236,7 +272,8 @@ public Fileset createFileset( | |
try { | ||
// formalize the path to avoid path without scheme, uri, authority, etc. | ||
filesetPath = formalizePath(filesetPath, hadoopConf); | ||
FileSystem fs = filesetPath.getFileSystem(hadoopConf); | ||
|
||
FileSystem fs = getFileSystem(filesetPath, hadoopConf); | ||
if (!fs.exists(filesetPath)) { | ||
if (!fs.mkdirs(filesetPath)) { | ||
throw new RuntimeException( | ||
|
@@ -339,7 +376,7 @@ public boolean dropFileset(NameIdentifier ident) { | |
|
||
// For managed fileset, we should delete the related files. | ||
if (filesetEntity.filesetType() == Fileset.Type.MANAGED) { | ||
FileSystem fs = filesetPath.getFileSystem(hadoopConf); | ||
FileSystem fs = getFileSystem(filesetPath, hadoopConf); | ||
if (fs.exists(filesetPath)) { | ||
if (!fs.delete(filesetPath, true)) { | ||
LOG.warn("Failed to delete fileset {} location {}", ident, filesetPath); | ||
|
@@ -459,7 +496,7 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str | |
Path schemaPath = getSchemaPath(ident.name(), properties); | ||
if (schemaPath != null) { | ||
try { | ||
FileSystem fs = schemaPath.getFileSystem(hadoopConf); | ||
FileSystem fs = getFileSystem(schemaPath, hadoopConf); | ||
if (!fs.exists(schemaPath)) { | ||
if (!fs.mkdirs(schemaPath)) { | ||
// Fail the operation when failed to create the schema path. | ||
|
@@ -577,8 +614,7 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty | |
if (schemaPath == null) { | ||
return false; | ||
} | ||
|
||
FileSystem fs = schemaPath.getFileSystem(hadoopConf); | ||
FileSystem fs = getFileSystem(schemaPath, hadoopConf); | ||
// Nothing to delete if the schema path does not exist. | ||
if (!fs.exists(schemaPath)) { | ||
return false; | ||
|
@@ -731,7 +767,7 @@ private boolean hasCallerContext() { | |
private boolean checkSingleFile(Fileset fileset) { | ||
try { | ||
Path locationPath = new Path(fileset.storageLocation()); | ||
return locationPath.getFileSystem(hadoopConf).getFileStatus(locationPath).isFile(); | ||
return getFileSystem(locationPath, hadoopConf).getFileStatus(locationPath).isFile(); | ||
} catch (FileNotFoundException e) { | ||
// We should always return false here, same with the logic in `FileSystem.isFile(Path f)`. | ||
return false; | ||
|
@@ -742,4 +778,27 @@ private boolean checkSingleFile(Fileset fileset) { | |
fileset.name()); | ||
} | ||
} | ||
|
||
static FileSystem getFileSystem(Path path, Configuration conf) throws IOException { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we don't have use Path to create a FileSystem, Configuration itself should be enough. |
||
if (path == null) { | ||
String defaultFS = conf.get("fs.defaultFS"); | ||
if (defaultFS != null) { | ||
path = new Path(defaultFS); | ||
} | ||
} | ||
|
||
String scheme; | ||
if (path == null || path.toUri().getScheme() == null) { | ||
scheme = "file"; | ||
} else { | ||
scheme = path.toUri().getScheme(); | ||
} | ||
|
||
FileSystemProvider provider = FILE_SYSTEM_PROVIDERS.get(scheme); | ||
if (provider == null) { | ||
throw new IllegalArgumentException("Unsupported scheme: " + scheme); | ||
} | ||
|
||
return provider.getFileSystem(conf, path); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,80 @@ | ||
/* | ||
* 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.gravitino.catalog.hadoop.fs; | ||
|
||
import java.io.IOException; | ||
import java.net.URI; | ||
import org.apache.commons.lang3.StringUtils; | ||
import org.apache.gravitino.catalog.hadoop.FileSystemProvider; | ||
import org.apache.hadoop.conf.Configuration; | ||
import org.apache.hadoop.fs.FileSystem; | ||
import org.apache.hadoop.fs.Path; | ||
|
||
public class HDFSFileSystemProvider implements FileSystemProvider { | ||
|
||
@Override | ||
public FileSystem getFileSystem(Configuration configuration, Path path) throws IOException { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @jerryshao |
||
Path fileSystemPath = path; | ||
if (fileSystemPath == null) { | ||
String pathString = configuration.get("fs.defaultFS"); | ||
if (StringUtils.isNotBlank(pathString)) { | ||
fileSystemPath = new Path(pathString); | ||
} | ||
} | ||
|
||
if (fileSystemPath == null) { | ||
throw new IllegalArgumentException("The path should be specified."); | ||
} | ||
|
||
URI uri = fileSystemPath.toUri(); | ||
if (uri.getScheme() != null && !uri.getScheme().equals("hdfs")) { | ||
throw new IllegalArgumentException("The path should be a HDFS path."); | ||
} | ||
|
||
// Should we call DistributedFileSystem to create file system instance explicitly? If we | ||
// explicitly create a HDFS file system here, we can't reuse the file system cache in the | ||
// FileSystem class. | ||
String impl = configuration.get("fs.hdfs.impl"); | ||
if (impl == null) { | ||
configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem"); | ||
} else { | ||
if (!impl.equals("org.apache.hadoop.hdfs.DistributedFileSystem")) { | ||
throw new IllegalArgumentException( | ||
"The HDFS file system implementation class should be 'org.apache.hadoop.hdfs.DistributedFileSystem'."); | ||
} | ||
} | ||
|
||
try { | ||
if (HDFSFileSystemProvider.class.getClassLoader().loadClass(configuration.get("fs.hdfs.impl")) | ||
== null) { | ||
throw new IllegalArgumentException( | ||
"The HDFS file system implementation class is not found."); | ||
} | ||
} catch (ClassNotFoundException e) { | ||
throw new IllegalArgumentException("The HDFS file system implementation class is not found."); | ||
} | ||
|
||
return FileSystem.get(uri, configuration); | ||
} | ||
|
||
@Override | ||
public String getScheme() { | ||
return "hdfs"; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,46 @@ | ||
/* | ||
* 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.gravitino.catalog.hadoop.fs; | ||
|
||
import java.io.IOException; | ||
import org.apache.commons.lang3.StringUtils; | ||
import org.apache.gravitino.catalog.hadoop.FileSystemProvider; | ||
import org.apache.hadoop.conf.Configuration; | ||
import org.apache.hadoop.fs.FileSystem; | ||
import org.apache.hadoop.fs.Path; | ||
|
||
public class LocalFileSystemProvider implements FileSystemProvider { | ||
|
||
@Override | ||
public FileSystem getFileSystem(Configuration configuration, Path path) throws IOException { | ||
Path fileSystemPath = path; | ||
if (fileSystemPath == null) { | ||
String pathString = configuration.get("fs.defaultFS"); | ||
if (StringUtils.isNotBlank(pathString)) { | ||
fileSystemPath = new Path(pathString); | ||
} | ||
} | ||
|
||
if (fileSystemPath == null) { | ||
fileSystemPath = new Path("file:///"); | ||
} | ||
|
||
return FileSystem.get(fileSystemPath.toUri(), configuration); | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@jerryshao
Please take a look at this PR, thanks.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we need to provide a Path here?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I noticed that most methods can get a FileSystem in https://github.com/apache/hadoop/blob/a9b7913d5689cae804fbd072a03cd2d2f771ab6a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L536-L571 do support specify the path explicitly, so I added it.
Indeed, we can also use configuration
fs.defaultFs
to replace it. If you feel it's an excessive value, I can remove it.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think the current interface design is highly dependent on the Hadoop's implementation, like
Configuration
andPath
, which is not a good design, also lose the flexibility to support some non-hadoop fs, a good interface should not rely on the third party libraries as possible as we can.I would be inclined to only rely on
FileSystem
, like: