[SUPPORT]Flink SQL No FileSystem for scheme s3

[ngk2009]

Tips before filing an issue

• Have you gone through our FAQs?
  yes

Describe the problem you faced

I used the Hudi0.10 version on the flink1.13.2 version. The flink storage has been successfully set to s3 access, and the storage access based on hdfs/alluxio is also normal, but there will be errors in the problem when the storage based on s3 is performed.

To Reproduce

Steps to reproduce the behavior:

flink set:
1.cd plugins/
2.mkdir s3-fs-hadoop
3.cd ..
4.cp opt/flink-s3-fs-hadoop-1.13.2.jar plugins/s3-fs-hadoop/
5.flink-conf.yaml
s3.access-key: xxx
s3.secret-key: xxx
s3.endpoint: xxx.xxx.xxx.xxx:7480
s3.path.style.access: true
s3.ssl.enabled: false
6.copy aws-java-sdk-bundle-1.11.134.jar/hadoop-aws-3.0.0-cdh6.3.0.jar to flink lib and set Hadoop class path:
export HADOOP_CLASSPATH=hadoop classpath
7.in Flink SQL Client
CREATE TABLE hudi_demo(
id BIGINT PRIMARY KEY NOT ENFORCED,
name STRING,
birthday TIMESTAMP(3),
ts TIMESTAMP(3),
partition VARCHAR(20)
) PARTITIONED BY (partition) WITH (
'connector' = 'hudi',
'table.type' = 'MERGE_ON_READ',
'path' = 's3://lakehouse/hudi/demo1/'
);
8.do a query
select * from hudi_demo;

Expected behavior

Can not execute a Query or Insert: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "s3"

Environment Description

• Hudi version : 0.10.0

• Spark version :N/A

• Flink version : 1.13.2/1.13.3

• Hive version :N/A

• Hadoop version :3.0.0-CDH6.3.0

• Storage (HDFS/S3/GCS..) :S3

• Running on Docker? (yes/no) :no
  Stacktrace

org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException: Could not start RpcEndpoint jobmanager_2.
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StoppedState.start(AkkaRpcActor.java:610) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:180) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.actor.Actor$class.aroundReceive(Actor.scala:517) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.actor.ActorCell.invoke(ActorCell.scala:561) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [flink-dist_2.11-1.13.2.jar:1.13.2]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [flink-dist_2.11-1.13.2.jar:1.13.2]
Caused by: org.apache.flink.runtime.jobmaster.JobMasterException: Could not start the JobMaster.
at org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:385) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStart(RpcEndpoint.java:181) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StoppedState.start(AkkaRpcActor.java:605) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
... 18 more
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to start the operator coordinators
at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.startAllOperatorCoordinators(DefaultOperatorCoordinatorHandler.java:90) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.SchedulerBase.startScheduling(SchedulerBase.java:592) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startScheduling(JobMaster.java:955) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startJobExecution(JobMaster.java:873) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:383) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStart(RpcEndpoint.java:181) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StoppedState.start(AkkaRpcActor.java:605) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
... 18 more
Caused by: org.apache.hudi.exception.HoodieIOException: Failed to get instance of org.apache.hadoop.fs.FileSystem
at org.apache.hudi.common.fs.FSUtils.getFs(FSUtils.java:104) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.util.StreamerUtil.tableExists(StreamerUtil.java:266) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.util.StreamerUtil.initTableIfNotExists(StreamerUtil.java:236) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.start(StreamWriteOperatorCoordinator.java:164) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.start(OperatorCoordinatorHolder.java:194) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.startAllOperatorCoordinators(DefaultOperatorCoordinatorHandler.java:85) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.SchedulerBase.startScheduling(SchedulerBase.java:592) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startScheduling(JobMaster.java:955) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startJobExecution(JobMaster.java:873) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:383) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStart(RpcEndpoint.java:181) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StoppedState.start(AkkaRpcActor.java:605) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
... 18 more
Caused by: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "s3"
at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3215) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3235) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:123) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3286) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3254) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:478) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]
at org.apache.hudi.common.fs.FSUtils.getFs(FSUtils.java:102) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.util.StreamerUtil.tableExists(StreamerUtil.java:266) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.util.StreamerUtil.initTableIfNotExists(StreamerUtil.java:236) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.start(StreamWriteOperatorCoordinator.java:164) ~[hudi-flink-bundle_2.11-0.10.0.jar:0.10.0]
at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.start(OperatorCoordinatorHolder.java:194) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.startAllOperatorCoordinators(DefaultOperatorCoordinatorHandler.java:85) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.SchedulerBase.startScheduling(SchedulerBase.java:592) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startScheduling(JobMaster.java:955) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.startJobExecution(JobMaster.java:873) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.onStart(JobMaster.java:383) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStart(RpcEndpoint.java:181) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StoppedState.start(AkkaRpcActor.java:605) ~[flink-dist_2.11-1.13.2.jar:1.13.2]

[danny0405]

you need to set up the s3 config in hadoop core-site file ? do you package the s3 package in the bundle jar ?

[ngk2009]

you need to set up the s3 config in hadoop core-site file ? do you package the s3 package in the bundle jar ?

Thank you for your reply, I did not find in the guide about flink setting core-site.xml, and I do not plan to use hadoop environment, just build data lake analysis based on S3 storage。
The bundle jar I found in hadoop jars library folder，Can't confirm whether it's packaged in。
Regarding whether Hadoop has a dependency on S3, I think it can be ruled out, because I use the s3 storage for the state.backend configured by flink and run it successfully on yarn.

[danny0405]

Flink use its own plugin to support filesystems other than HDFS. Hudi adapter to different DFS by extending the FileSystem interface directly.

[york-yu-ctw]

I have some tips

1. put flink-s3-fs-hadoop into /opt/flink/lib
2. add hadoop-hdfs-client, hadoop-aws, hadoop-mapreduce-client-core into /opt/flink/lib as well
3. if there is aws credential issue, try to set up core-site.xml (I had this issue when using hudi 0.9.0 and flink 1.12.2)

[ngk2009]

Flink use its own plugin to support filesystems other than HDFS. Hudi adapter to different DFS by extending the FileSystem interface directly.

How to solve it？thanks

[ngk2009]

I have some tips

1. put `flink-s3-fs-hadoop` into `/opt/flink/lib`

2. add `hadoop-hdfs-client`, `hadoop-aws`, `hadoop-mapreduce-client-core` into  `/opt/flink/lib` as well

3. if there is aws credential issue, try to set up core-site.xml (I had this issue when using hudi 0.9.0 and flink 1.12.2)

Unfortunately, I tried your method but did not succeed。Copy those jar packages to the /opt/flink/lib and $FLINK_HOME/lib directories at the same time, and then restart the yarn session :
ERROR org.apache.flink.yarn.cli.FlinkYarnSessionCli [] - Error while running the Flink session.
java.lang.NoSuchFieldError: IGNORE_CLIENT_LOCALITY

This may be caused by the hadoop version that flink depends on. I try to recompile flink-fs-hadoop. If hudi does not rely on hadoop's filesystem, just use s3's sdk, the dependence is too heavy

[danny0405]

Flink use its own plugin to support filesystems other than HDFS. Hudi adapter to different DFS by extending the FileSystem interface directly.

How to solve it？thanks

Hudi did depend on the hadoop FileSystem interface, what we need to do is adding the aws s3 FileSystem impl codes in the classpath, and it's specific configuration should also be configured in hadoop Configuration, you can reference the StreamerUtil.getHadoopConf how we fetch the hadoop configuration in the flink pipeline.

[ngk2009]

Flink use its own plugin to support filesystems other than HDFS. Hudi adapter to different DFS by extending the FileSystem interface directly.

How to solve it？thanks

Hudi did depend on the hadoop FileSystem interface, what we need to do is adding the aws s3 FileSystem impl codes in the classpath, and it's specific configuration should also be configured in hadoop Configuration, you can reference the StreamerUtil.getHadoopConf how we fetch the hadoop configuration in the flink pipeline.

hi,I found you changed the class loader in file (#4042) org.apache.hudi.sink.StreamWriteOperatorCoordinator：Thread.currentThread().setContextClassLoader(getClass().getClassLoader()),Will the classpath in the hudi-bundle jar be used? Will the classes under flink/lib fail to load and cause the S3 schema not to be found? so which "the aws s3 FileSystem impl codes in the classpath" package to be add in flink lib?
current in flink lib jars are:
hadoop-aws-3.0.0-cdh6.3.0.jar
hadoop-common-3.0.0-cdh6.3.0.jar
hadoop-hdfs-client-3.0.0-cdh6.3.0.jar
hadoop-mapreduce-client-core-3.0.0-cdh6.3.0.jar
flink-s3-fs-hadoop-1.13.3.jar
aws-java-sdk-s3-1.11.836.jar
hudi-aws-0.10.0.jar
hudi-flink-bundle_2.11-0.10.0.jar

[ngk2009]

S3:// cannot be used now, and s3a:// can be used. In addition, s3-fs-hadoop under flink will always appear in the INITIALIZING state when the task is submitted. In addition, the dependency of hadoop and core-site.xml must be required. ,This dependency is too heavy, I switched to using alluxio:// to write data to S3 UFS 。
Thanks for help。