[FLINK-27803] Unable to write to s3 with Hudi format via Flink - Scala - ASF JIRA

XML

Word

Printable

JSON

Details

Type: Bug
Status: Closed
Priority: Major
Resolution: Invalid
Affects Version/s: 1.14.4
Fix Version/s: None
Component/s: Connectors / Hadoop Compatibility
Labels:
- HUDI-bug
Environment:
Hide

SBT:
===

libraryDependencies ++= Seq(
"org.apache.flink" %% "flink-scala" % "1.10.0" % "provided",
"org.apache.flink" %% "flink-streaming-scala" % "1.10.0" % "provided",
"com.amazonaws" % "aws-java-sdk-s3" % "1.11.563" % "provided",
"org.apache.flink" %% "flink-parquet" % "1.14.4" % "provided",
"org.apache.hadoop" % "hadoop-aws" % "3.2.2" % "provided",
"org.apache.parquet" % "parquet-avro" % "1.10.1",
"com.typesafe.scala-logging" %% "scala-logging" % "3.8.0",

"org.apache.flink" %% "flink-table-api-scala" % "1.14.4",
// "org.apache.flink" % "flink-table-api-java" % "1.14.4",
"org.apache.flink" %% "flink-table-api-scala-bridge" % "1.14.4" % "provided",
"org.apache.flink" %% "flink-table-planner-blink" % "1.13.5" % "provided",
"org.apache.flink" %% "flink-table-planner" % "1.14.4" % "provided",
"org.apache.flink" % "flink-table-common" % "1.14.4" % "provided",
"org.apache.flink" %% "flink-clients" % "1.14.4" % "provided",
"org.apache.hudi" % "hudi-flink-bundle_2.11" % "0.10.1" % "provided",

"org.apache.hudi" % "hudi-common" % "0.11.0" % "provided",
"org.apache.hudi" % "hudi-client-common" % "0.11.0" % "provided",
"org.apache.hudi" % "hudi-flink-client" % "0.11.0" % "provided",
"org.apache.hudi" % "hudi-sync-common" % "0.11.0" % "provided",
// "com.amazonaws" % "aws-java-sdk-bundle" % "1.11.563" % "provided",
"org.apache.hadoop" % "hadoop-common" % "3.2.2" % "provided",
"org.apache.hadoop" % "hadoop-client" % "3.2.2" % "provided",
"org.apache.parquet" % "parquet-hadoop" % "1.12.2" % "provided",
"org.apache.parquet" % "parquet-hadoop-bundle" % "1.12.2" % "provided",
"org.apache.flink" % "flink-s3-fs-hadoop" % "1.14.4" % "provided"
)

assemblyMergeStrategy in assembly :=
{ casePathList("META-INF", xs @ _*) => MergeStrategy.discard case x => MergeStrategy.first }

Docker file:
=========
FROM flink:1.14.4-scala_2.12

ENV FLINK_HADOOP_CONF /hadoop/conf

RUN curl https://archive.apache.org/dist/flink/flink-1.14.4/flink-1.14.4-bin-scala_2.12.tgz -o /opt/flink-1.14.4-bin-scala_2.12.tgz
RUN tar -xzf /opt/flink-1.14.4-bin-scala_2.12.tgz
RUN mv flink-1.14.4 /opt

RUN rm -rf /opt/flink/*
RUN rm -rf /opt/flink/plugins/*
RUN rm -rf /opt/flink/lib/*

RUN cp -R /opt/flink-1.14.4/* /opt/flink

RUN ls -ltra /opt/flink

RUN mv -v /opt/flink/opt/flink-python*.jar /opt/flink/lib/
RUN mkdir -p /opt/flink/plugins/s3-fs-presto /opt/flink/plugins/s3-fs-hadoop
RUN mv ~~v /opt/flink/opt/flink-s3-fs-presto~~*.jar /opt/flink/plugins/s3-fs-presto
RUN mv ~~v /opt/flink/opt/flink-s3-fs-hadoop~~*.jar /opt/flink/plugins/s3-fs-hadoop
RUN mv ~~v /opt/flink/opt/flink~~*.jar /opt/flink/plugins/

RUN wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar -P /opt/flink/lib/
RUN wget https://repo1.maven.org/maven2/org/apache/hudi/hudi-flink-bundle_2.12/0.10.1/hudi-flink-bundle_2.12-0.10.1.jar -P /opt/flink/lib/
RUN wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-s3/1.11.563/aws-java-sdk-s3-1.11.563.jar -P /opt/flink/lib/
RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.2.2/hadoop-aws-3.2.2.jar -P /opt/flink/lib/

RUN wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.563/aws-java-sdk-bundle-1.11.563.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client/3.2.2/hadoop-client-3.2.2.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-common/3.2.2/hadoop-common-3.2.2.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-table-api-scala_2.12/1.14.4/flink-table-api-scala_2.12-1.14.4.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/com/google/guava/guava/31.0-jre/guava-31.0-jre.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-hadoop/1.14.4/flink-s3-fs-hadoop-1.14.4.jar -P /opt/flink/lib/
RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs-client/3.2.2/hadoop-hdfs-client-3.2.2.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-mapreduce-client-core/3.3.2/hadoop-mapreduce-client-core-3.3.2.jar -P /opt/flink/lib

For Parquet
RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-parquet_2.12/1.14.4/flink-parquet_2.12-1.14.4.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/parquet/parquet-hadoop/1.12.2/parquet-hadoop-1.12.2.jar -P /opt/flink/lib
RUN wget https://repo1.maven.org/maven2/org/apache/parquet/parquet-hadoop-bundle/1.12.2/parquet-hadoop-bundle-1.12.2.jar -P /opt/flink/lib

RUN chown -R flink:flink /opt/flink/plugins/
RUN chown -R flink:flink /opt/flink/lib/
Show
SBT: === libraryDependencies ++= Seq( "org.apache.flink" %% "flink-scala" % "1.10.0" % "provided", "org.apache.flink" %% "flink-streaming-scala" % "1.10.0" % "provided", "com.amazonaws" % "aws-java-sdk-s3" % "1.11.563" % "provided", "org.apache.flink" %% "flink-parquet" % "1.14.4" % "provided", "org.apache.hadoop" % "hadoop-aws" % "3.2.2" % "provided", "org.apache.parquet" % "parquet-avro" % "1.10.1", "com.typesafe.scala-logging" %% "scala-logging" % "3.8.0", "org.apache.flink" %% "flink-table-api-scala" % "1.14.4", // "org.apache.flink" % "flink-table-api-java" % "1.14.4", "org.apache.flink" %% "flink-table-api-scala-bridge" % "1.14.4" % "provided", "org.apache.flink" %% "flink-table-planner-blink" % "1.13.5" % "provided", "org.apache.flink" %% "flink-table-planner" % "1.14.4" % "provided", "org.apache.flink" % "flink-table-common" % "1.14.4" % "provided", "org.apache.flink" %% "flink-clients" % "1.14.4" % "provided", "org.apache.hudi" % "hudi-flink-bundle_2.11" % "0.10.1" % "provided", "org.apache.hudi" % "hudi-common" % "0.11.0" % "provided", "org.apache.hudi" % "hudi-client-common" % "0.11.0" % "provided", "org.apache.hudi" % "hudi-flink-client" % "0.11.0" % "provided", "org.apache.hudi" % "hudi-sync-common" % "0.11.0" % "provided", // "com.amazonaws" % "aws-java-sdk-bundle" % "1.11.563" % "provided", "org.apache.hadoop" % "hadoop-common" % "3.2.2" % "provided", "org.apache.hadoop" % "hadoop-client" % "3.2.2" % "provided", "org.apache.parquet" % "parquet-hadoop" % "1.12.2" % "provided", "org.apache.parquet" % "parquet-hadoop-bundle" % "1.12.2" % "provided", "org.apache.flink" % "flink-s3-fs-hadoop" % "1.14.4" % "provided" ) assemblyMergeStrategy in assembly := { casePathList("META-INF", xs @ _*) => MergeStrategy.discard case x => MergeStrategy.first } Docker file: ========= FROM flink:1.14.4-scala_2.12 ENV FLINK_HADOOP_CONF /hadoop/conf RUN curl https://archive.apache.org/dist/flink/flink-1.14.4/flink-1.14.4-bin-scala_2.12.tgz -o /opt/flink-1.14.4-bin-scala_2.12.tgz RUN tar -xzf /opt/flink-1.14.4-bin-scala_2.12.tgz RUN mv flink-1.14.4 /opt RUN rm -rf /opt/flink/* RUN rm -rf /opt/flink/plugins/* RUN rm -rf /opt/flink/lib/* RUN cp -R /opt/flink-1.14.4/* /opt/flink RUN ls -ltra /opt/flink RUN mv -v /opt/flink/opt/flink-python*.jar /opt/flink/lib/ RUN mkdir -p /opt/flink/plugins/s3-fs-presto /opt/flink/plugins/s3-fs-hadoop RUN mv v /opt/flink/opt/flink-s3-fs-presto *.jar /opt/flink/plugins/s3-fs-presto RUN mv v /opt/flink/opt/flink-s3-fs-hadoop *.jar /opt/flink/plugins/s3-fs-hadoop RUN mv v /opt/flink/opt/flink *.jar /opt/flink/plugins/ RUN wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar -P /opt/flink/lib/ RUN wget https://repo1.maven.org/maven2/org/apache/hudi/hudi-flink-bundle_2.12/0.10.1/hudi-flink-bundle_2.12-0.10.1.jar -P /opt/flink/lib/ RUN wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-s3/1.11.563/aws-java-sdk-s3-1.11.563.jar -P /opt/flink/lib/ RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.2.2/hadoop-aws-3.2.2.jar -P /opt/flink/lib/ RUN wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.563/aws-java-sdk-bundle-1.11.563.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client/3.2.2/hadoop-client-3.2.2.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-common/3.2.2/hadoop-common-3.2.2.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-table-api-scala_2.12/1.14.4/flink-table-api-scala_2.12-1.14.4.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/com/google/guava/guava/31.0-jre/guava-31.0-jre.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-hadoop/1.14.4/flink-s3-fs-hadoop-1.14.4.jar -P /opt/flink/lib/ RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs-client/3.2.2/hadoop-hdfs-client-3.2.2.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-mapreduce-client-core/3.3.2/hadoop-mapreduce-client-core-3.3.2.jar -P /opt/flink/lib For Parquet RUN wget https://repo1.maven.org/maven2/org/apache/flink/flink-parquet_2.12/1.14.4/flink-parquet_2.12-1.14.4.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/parquet/parquet-hadoop/1.12.2/parquet-hadoop-1.12.2.jar -P /opt/flink/lib RUN wget https://repo1.maven.org/maven2/org/apache/parquet/parquet-hadoop-bundle/1.12.2/parquet-hadoop-bundle-1.12.2.jar -P /opt/flink/lib RUN chown -R flink:flink /opt/flink/plugins/ RUN chown -R flink:flink /opt/flink/lib/

Flags:

Important
Language:
- scala

Description

Getting this error , when writing to S3 as Hudi file format via Flink:

java.nio.file.AccessDeniedException: data: org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException: No AWS Credentials provided by SimpleAWSCredentialsProvider EnvironmentVariableCredentialsProvider InstanceProfileCredentialsProvider : com.amazonaws.SdkClientException: Failed to connect to service endpoint:

Tried updating the core-site.xml in the class path in the libs
All the Hadoop and awssdk jars are in the lib folder

Stacktrace:

13:45:27.026 [main] ERROR org.apache.flink.core.fs.FileSystem - Failed to load a file system via services

java.util.ServiceConfigurationError: org.apache.flink.core.fs.FileSystemFactory: Provider org.apache.flink.fs.s3presto.S3FileSystemFactory could not be instantiated

at java.util.ServiceLoader.fail(ServiceLoader.java:232) ~[?:1.8.0_332]

at java.util.ServiceLoader.access$100(ServiceLoader.java:185) ~[?:1.8.0_332]

at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:384) ~[?:1.8.0_332]

at java.util.ServiceLoader$LazyIterator.next(ServiceLoader.java:404) ~[?:1.8.0_332]

at java.util.ServiceLoader$1.next(ServiceLoader.java:480) ~[?:1.8.0_332]

at org.apache.flink.core.plugin.PluginLoader$ContextClassLoaderSettingIterator.next(PluginLoader.java:136) ~[flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.shaded.guava30.com.google.common.collect.Iterators$ConcatenatedIterator.next(Iterators.java:1364) ~[flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.shaded.guava30.com.google.common.collect.TransformedIterator.next(TransformedIterator.java:47) ~[flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.core.fs.FileSystem.addAllFactoriesToList(FileSystem.java:1079) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.core.fs.FileSystem.loadFileSystemFactories(FileSystem.java:1060) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.core.fs.FileSystem.initialize(FileSystem.java:340) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.configureFileSystems(ClusterEntrypoint.java:229) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:185) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:617) [flink-dist_2.12-1.14.4.jar:1.14.4]

at org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint.main(StandaloneSessionClusterEntrypoint.java:59) [flink-dist_2.12-1.14.4.jar:1.14.4]

Caused by: java.lang.LinkageError: loader constraint violation: when resolving overridden method "org.apache.flink.fs.s3presto.S3FileSystemFactory.createHadoopFileSystem()Lorg/apache/hadoop/fs/FileSystem;" the class loader (instance of org/apache/flink/core/plugin/PluginLoader$PluginClassLoader) of the current class, org/apache/flink/fs/s3presto/S3FileSystemFactory, and its superclass loader (instance of sun/misc/Launcher$AppClassLoader), have different Class objects for the type org/apache/hadoop/fs/FileSystem used in the signature

at java.lang.Class.getDeclaredConstructors0(Native Method) ~[?:1.8.0_332]

at java.lang.Class.privateGetDeclaredConstructors(Class.java:2671) ~[?:1.8.0_332]

at java.lang.Class.getConstructor0(Class.java:3075) ~[?:1.8.0_332]

at java.lang.Class.newInstance(Class.java:412) ~[?:1.8.0_332]

at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:380) ~[?:1.8.0_332]

... 12 more

13:45:27.032 [main] ERROR org.apache.flink.core.fs.FileSystem - Failed to load a file system via services

java.util.ServiceConfigurationError: org.apache.flink.core.fs.FileSystemFactory: Provider org.apache.flink.fs.s3presto.S3PFileSystemFactory could not be instantiated