[FLINK-33569] Could not deploy yarn-application when using yarn over s3a filesystem.

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Duplicate
• Affects Version/s: 1.18.0, 1.17.1
• Component/s: Deployment / YARN

Environment

Env:

• OS: ArchLinux kernel:6.6.1 AMD64
• Flink: 1.17.1
• Hadoop: 3.3.6
• Minio: 2023-11-15

Settings

hadoop core-site.xml:

 

  <property>
    <name>fs.defaultFS</name>    
    <value>s3a://hadoop</value>
  </property>
  <property>
    <name>fs.s3a.path.style.access</name>
    <value>true</value>
  </property>
  <!-- minio username -->
  <property>
    <name>fs.s3a.access.key</name>
    <value>admin</value>
  </property>
  <!-- minio password -->
  <property>
    <name>fs.s3a.secret.key</name>
    <value>password</value>
  </property>
  <!-- minio endpoint -->
  <property>
    <name>fs.s3a.endpoint</name>
    <value>http://localhost:9000</value>
  </property>
  <property>
    <name>fs.s3a.connection.establish.timeout</name>
    <value>5000</value>
  </property>
  <property>
    <name>fs.s3a.multipart.size</name>
    <value>512M</value>
  </property>
  <property>
    <name>fs.s3a.impl</name>
    <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
  </property>
  <property>
    <name>fs.AbstractFileSystem.s3a.impl</name>
    <value>org.apache.hadoop.fs.s3a.S3A</value>
  </property>
  <!-- S3 end -->

Flink run command:

./bin/flink run-application -t yarn-application ./examples/streaming/TopSpeedWindowing.jar

 

 

• Assignee: Unassigned
• Reporter: Bodong Liu
• Created: 16/Nov/23 08:49
• Updated: 20/Nov/23 14:49
• Resolved: 20/Nov/23 14:47

Description

 

I now use the `yarn-application` mode to deploy Flink. I found that when I set Hadoop's storage to the s3a file system, Flink could not submit tasks to Yarn.
The error is reported as follows:

------------------------------------------------------------

 The program finished with the following exception:



org.apache.flink.client.deployment.ClusterDeploymentException: Couldn't deploy Yarn Application Cluster

        at org.apache.flink.yarn.YarnClusterDescriptor.deployApplicationCluster(YarnClusterDescriptor.java:481)

        at org.apache.flink.client.deployment.application.cli.ApplicationClusterDeployer.run(ApplicationClusterDeployer.java:67)

        at org.apache.flink.client.cli.CliFrontend.runApplication(CliFrontend.java:212)

        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1098)

        at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)

        at java.security.AccessController.doPrivileged(Native Method)

        at javax.security.auth.Subject.doAs(Subject.java:422)

        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1899)

        at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)

        at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)

        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)

Caused by: org.apache.hadoop.fs.PathIOException: `Cannot get relative path for URI:file:///tmp/application_1700122774429_0001-flink-conf.yaml5526160496134930395.tmp': Input/output error

        at org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation.getFinalPath(CopyFromLocalOperation.java:360)

        at org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation.uploadSourceFromFS(CopyFromLocalOperation.java:222)

        at org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation.execute(CopyFromLocalOperation.java:169)

        at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$copyFromLocalFile$26(S3AFileSystem.java:3854)

        at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration(IOStatisticsBinding.java:547)

        at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.lambda$trackDurationOfOperation$5(IOStatisticsBinding.java:528)

        at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration(IOStatisticsBinding.java:449)

        at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2480)

        at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2499)

        at org.apache.hadoop.fs.s3a.S3AFileSystem.copyFromLocalFile(S3AFileSystem.java:3847)

        at org.apache.flink.yarn.YarnApplicationFileUploader.copyToRemoteApplicationDir(YarnApplicationFileUploader.java:397)

        at org.apache.flink.yarn.YarnApplicationFileUploader.uploadLocalFileToRemote(YarnApplicationFileUploader.java:202)

        at org.apache.flink.yarn.YarnApplicationFileUploader.registerSingleLocalResource(YarnApplicationFileUploader.java:181)

        at org.apache.flink.yarn.YarnClusterDescriptor.startAppMaster(YarnClusterDescriptor.java:1050)

        at org.apache.flink.yarn.YarnClusterDescriptor.deployInternal(YarnClusterDescriptor.java:626)

        at org.apache.flink.yarn.YarnClusterDescriptor.deployApplicationCluster(YarnClusterDescriptor.java:474)

        ... 10 more


 

I found by looking through the source code and debugging that when Hadoop uses the s3a file system, uploading and downloading files must use URIs with `scheme` to build path parameters.

In the `org.apache.flink.yarn.YarnClusterDescriptor` class, when uploading a temporarily generated `yaml` configuration file, the absolute path of the file is used instead of the URI as the path construction parameter, but other file upload and download behaviors They all use URI as the path parameter.

This is the reason for the error reported above.

Issue Links

• duplicates: FLINK-33472 Solve the problem that the temporary file of flink-conf.yaml in S3AFileSystem cannot be uploaded