[
https://issues.apache.org/jira/browse/FLINK-38146?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
jonasjc updated FLINK-38146:
----------------------------
Description:
*Describtion:*
For create filesystem support different config , support
Configuration param when get filesystem:
{code:java}
public interface FileSystemFactory extends Plugin {
/** Gets the scheme of the file system created by this factory. */
String getScheme();
/**
* Creates a new file system for the given file system URI. The URI
describes the type of file
* system (via its scheme) and optionally the authority (for example the
host) of the file
* system.
*
* @param fsUri The URI that describes the file system.
* @return A new instance of the specified file system.
* @throws IOException Thrown if the file system could not be instantiated.
*/
FileSystem create(URI fsUri) throws IOException;
// New support interface
default FileSystem create(URI fsUri, Configuration configuration) throws
IOException {
throw new UnsupportedOperationException();
}
}{code}
*Why ?*
When need to create filesystem use catalog option or table option,we need to
prefer use option config rather than default config.Such as one implement of
Filesystem org.apache.flink.fs.osshadoop.OSSFileSystemFactory#create.
Old code :
{code:java}
@Override
public FileSystem create(URI fsUri) throws IOException {
this.hadoopConfig = getHadoopConfiguration();
final String scheme = fsUri.getScheme();
final String authority = fsUri.getAuthority();
if (scheme == null && authority == null) {
fsUri = org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
} else if (scheme != null && authority == null) {
URI defaultUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
if (scheme.equals(defaultUri.getScheme()) && defaultUri.getAuthority()
!= null) {
fsUri = defaultUri;
}
}
final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
fs.initialize(fsUri, hadoopConfig);
...
}{code}
We can only use config by getHadoopConfiguration,if i have multi
oss.endpoint、oss.
accessKeyId e.g. Now, is not support.
So,we need support config param get from connector catalog or table option.
New code:
{code:java}
@Override
public FileSystem create(URI fsUri, Configuration configuration) throws
IOException {
this.hadoopConfig = getHadoopConfiguration();
org.apache.hadoop.conf.Configuration copyHadoopConfig =
new org.apache.hadoop.conf.Configuration();
hadoopConfig.forEach(e -> copyHadoopConfig.set(e.getKey(), e.getValue()));
configuration.toMap().forEach(copyHadoopConfig::set);
final String scheme = fsUri.getScheme();
final String authority = fsUri.getAuthority();
if (scheme == null && authority == null) {
fsUri = org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
} else if (scheme != null && authority == null) {
URI defaultUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
if (scheme.equals(defaultUri.getScheme()) && defaultUri.getAuthority()
!= null) {
fsUri = defaultUri;
}
}
final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
fs.initialize(fsUri, copyHadoopConfig);
...
}{code}
was:
*Describtion:*
For create filesystem support different config , support
Configuration param when get filesystem:
{code:java}
public interface FileSystemFactory extends Plugin {
/** Gets the scheme of the file system created by this factory. */
String getScheme();
/**
* Creates a new file system for the given file system URI. The URI
describes the type of file
* system (via its scheme) and optionally the authority (for example the
host) of the file
* system.
*
* @param fsUri The URI that describes the file system.
* @return A new instance of the specified file system.
* @throws IOException Thrown if the file system could not be instantiated.
*/
FileSystem create(URI fsUri) throws IOException;
// New support interface
default FileSystem create(URI fsUri, Configuration configuration) throws
IOException {
throw new UnsupportedOperationException();
}
}{code}
*Why ?*
When need to create filesystem use catalog option or table option,we need to
prefer use option config rather than default config,such as implement of
org.apache.flink.fs.osshadoop.OSSFileSystemFactory#create.
Old code :
{code:java}
@Override
public FileSystem create(URI fsUri) throws IOException {
this.hadoopConfig = getHadoopConfiguration();
final String scheme = fsUri.getScheme();
final String authority = fsUri.getAuthority();
if (scheme == null && authority == null) {
fsUri = org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
} else if (scheme != null && authority == null) {
URI defaultUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
if (scheme.equals(defaultUri.getScheme()) && defaultUri.getAuthority()
!= null) {
fsUri = defaultUri;
}
}
final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
fs.initialize(fsUri, hadoopConfig);
...
}{code}
We can only use config by getHadoopConfiguration,if i have multi
oss.endpoint、oss.
accessKeyId and e.g. Now, is not support.
So,we need support config param get from connector catalog or table option.
New code:
{code:java}
@Override
public FileSystem create(URI fsUri, Configuration configuration) throws
IOException {
this.hadoopConfig = getHadoopConfiguration();
org.apache.hadoop.conf.Configuration copyHadoopConfig =
new org.apache.hadoop.conf.Configuration();
hadoopConfig.forEach(e -> copyHadoopConfig.set(e.getKey(), e.getValue()));
configuration.toMap().forEach(copyHadoopConfig::set);
final String scheme = fsUri.getScheme();
final String authority = fsUri.getAuthority();
if (scheme == null && authority == null) {
fsUri = org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
} else if (scheme != null && authority == null) {
URI defaultUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
if (scheme.equals(defaultUri.getScheme()) && defaultUri.getAuthority()
!= null) {
fsUri = defaultUri;
}
}
final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
fs.initialize(fsUri, copyHadoopConfig);
...
}{code}
> support config for create filesystem
> ------------------------------------
>
> Key: FLINK-38146
> URL: https://issues.apache.org/jira/browse/FLINK-38146
> Project: Flink
> Issue Type: Improvement
> Components: API / Core
> Reporter: jonasjc
> Priority: Major
>
> *Describtion:*
> For create filesystem support different config , support
> Configuration param when get filesystem:
> {code:java}
> public interface FileSystemFactory extends Plugin {
> /** Gets the scheme of the file system created by this factory. */
> String getScheme();
> /**
> * Creates a new file system for the given file system URI. The URI
> describes the type of file
> * system (via its scheme) and optionally the authority (for example the
> host) of the file
> * system.
> *
> * @param fsUri The URI that describes the file system.
> * @return A new instance of the specified file system.
> * @throws IOException Thrown if the file system could not be
> instantiated.
> */
> FileSystem create(URI fsUri) throws IOException;
> // New support interface
> default FileSystem create(URI fsUri, Configuration configuration) throws
> IOException {
> throw new UnsupportedOperationException();
> }
> }{code}
> *Why ?*
> When need to create filesystem use catalog option or table option,we need to
> prefer use option config rather than default config.Such as one implement of
> Filesystem org.apache.flink.fs.osshadoop.OSSFileSystemFactory#create.
>
> Old code :
> {code:java}
> @Override
> public FileSystem create(URI fsUri) throws IOException {
> this.hadoopConfig = getHadoopConfiguration();
> final String scheme = fsUri.getScheme();
> final String authority = fsUri.getAuthority();
> if (scheme == null && authority == null) {
> fsUri = org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
> } else if (scheme != null && authority == null) {
> URI defaultUri =
> org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
> if (scheme.equals(defaultUri.getScheme()) &&
> defaultUri.getAuthority() != null) {
> fsUri = defaultUri;
> }
> }
> final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
> fs.initialize(fsUri, hadoopConfig);
> ...
> }{code}
> We can only use config by getHadoopConfiguration,if i have multi
> oss.endpoint、oss.
> accessKeyId e.g. Now, is not support.
> So,we need support config param get from connector catalog or table option.
>
> New code:
> {code:java}
> @Override
> public FileSystem create(URI fsUri, Configuration configuration) throws
> IOException {
> this.hadoopConfig = getHadoopConfiguration();
> org.apache.hadoop.conf.Configuration copyHadoopConfig =
> new org.apache.hadoop.conf.Configuration();
> hadoopConfig.forEach(e -> copyHadoopConfig.set(e.getKey(), e.getValue()));
> configuration.toMap().forEach(copyHadoopConfig::set);
> final String scheme = fsUri.getScheme();
> final String authority = fsUri.getAuthority();
> if (scheme == null && authority == null) {
> fsUri =
> org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
> } else if (scheme != null && authority == null) {
> URI defaultUri =
> org.apache.hadoop.fs.FileSystem.getDefaultUri(copyHadoopConfig);
> if (scheme.equals(defaultUri.getScheme()) &&
> defaultUri.getAuthority() != null) {
> fsUri = defaultUri;
> }
> }
> final AliyunOSSFileSystem fs = new AliyunOSSFileSystem();
> fs.initialize(fsUri, copyHadoopConfig);
> ...
> }{code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
