https://flink.apache.org/ logo
Join Slack
Powered by
Hi all, I am trying to use the s3 plugin in flink ...
# troubleshooting
a
Hi all, I am trying to use the s3 plugin in flink 1.16. I have deployed the standalone kubernetes cluster in session mode using a custom dockerfile as follows:
Copy code
FROM apache/flink:1.16.0-scala_2.12

RUN mkdir -p /opt/flink/plugins/flink-s3-fs-hadoop
RUN mv /opt/flink/opt/flink-s3-fs-hadoop-*.jar /opt/flink/plugins/flink-s3-fs-hadoop/.
When I try to submit a job (using the new delta lake read connector) - here is some of the code to get an idea:
Copy code
DeltaSource<RowData> deviceDeltaTableStream = DeltaSource
      .forContinuousRowData(
            new Path("<s3a://my-delta-table-location>"),
            new Configuration())
      .columnNames("id", "state")
      .build();
Since the delta lake file is stored in s3 - I need to be able to access it from there. But when I submit this job - I get this error
Copy code
org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application.
	at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:113)
	at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.util.concurrent.CompletionException: org.apache.flink.util.FlinkRuntimeException: Could not execute application.
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
	... 2 more
Caused by: org.apache.flink.util.FlinkRuntimeException: Could not execute application.
	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:88)
	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)
	at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:107)
	... 2 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
	... 4 more
Caused by: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2592)
	at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3320)
	at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3352)
	at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:124)
	at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3403)
	at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3371)
	at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:477)
	at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361)
	at io.delta.standalone.internal.DeltaLogImpl$.apply(DeltaLogImpl.scala:260)
	at io.delta.standalone.internal.DeltaLogImpl$.forTable(DeltaLogImpl.scala:241)
	at io.delta.standalone.internal.DeltaLogImpl.forTable(DeltaLogImpl.scala)
	at io.delta.standalone.DeltaLog.forTable(DeltaLog.java:164)
What could be going wrong here? Also, attaching jobmanager logs in thread.
Jobmanager logs:
Copy code
2023-03-28 05:45:30,907 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - --------------------------------------------------------------------------------
2023-03-28 05:45:30,909 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Registered UNIX signal handlers for [TERM, HUP, INT]
2023-03-28 05:45:30,921 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: blob.server.port, 6124
2023-03-28 05:45:30,921 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.memory.process.size, 1728m
2023-03-28 05:45:30,921 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: queryable-state.proxy.ports, 6125
2023-03-28 05:45:30,921 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: parallelism.default, 2
2023-03-28 05:45:30,922 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.address, flink-jobmanager
2023-03-28 05:45:30,922 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.numberOfTaskSlots, 2
2023-03-28 05:45:30,922 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.memory.process.size, 1600m
2023-03-28 05:45:30,922 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.port, 6123
2023-03-28 05:45:30,923 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.rpc.port, 6122
2023-03-28 05:45:30,923 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: s3.path.style.access, true
2023-03-28 05:45:30,923 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading dynamic configuration property: jobmanager.memory.off-heap.size, 134217728b
2023-03-28 05:45:30,923 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading dynamic configuration property: jobmanager.memory.jvm-overhead.min, 201326592b
2023-03-28 05:45:30,923 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading dynamic configuration property: jobmanager.memory.jvm-metaspace.size, 268435456b
2023-03-28 05:45:30,924 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading dynamic configuration property: jobmanager.memory.heap.size, 1073741824b
2023-03-28 05:45:30,924 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading dynamic configuration property: jobmanager.memory.jvm-overhead.max, 201326592b
2023-03-28 05:45:30,949 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Starting StandaloneSessionClusterEntrypoint.
2023-03-28 05:45:30,987 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Install default filesystem.
2023-03-28 05:45:30,991 INFO  org.apache.flink.core.fs.FileSystem                          [] - Hadoop is not in the classpath/dependencies. The extended set of supported File Systems via Hadoop is not available.
2023-03-28 05:45:31,107 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Install security context.
2023-03-28 05:45:31,123 INFO  org.apache.flink.runtime.security.modules.HadoopModuleFactory [] - Cannot create Hadoop Security Module because Hadoop cannot be found in the Classpath.
2023-03-28 05:45:31,135 INFO  org.apache.flink.runtime.security.modules.JaasModule         [] - Jaas file will be created as /tmp/jaas-16277040840484793741.conf.
2023-03-28 05:45:31,143 INFO  org.apache.flink.runtime.security.contexts.HadoopSecurityContextFactory [] - Cannot install HadoopSecurityContext because Hadoop cannot be found in the Classpath.
2023-03-28 05:45:31,145 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Initializing cluster services.
2023-03-28 05:45:31,152 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Using working directory: WorkingDirectory(/tmp/jm_ba3e9448356424b7c239a69bc9fdf63c).
2023-03-28 05:45:31,692 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils        [] - Trying to start actor system, external address flink-jobmanager:6123, bind address 0.0.0.0:6123.
2023-03-28 05:45:32,644 INFO  akka.event.slf4j.Slf4jLogger                                 [] - Slf4jLogger started
2023-03-28 05:45:32,685 INFO  akka.remote.RemoteActorRefProvider                           [] - Akka Cluster not in use - enabling unsafe features anyway because `akka.remote.use-unsafe-remote-features-outside-cluster` has been enabled.
2023-03-28 05:45:32,692 INFO  akka.remote.Remoting                                         [] - Starting remoting
2023-03-28 05:45:32,911 INFO  akka.remote.Remoting                                         [] - Remoting started; listening on addresses :[<akka.tcp://flink@flink-jobmanager:6123>]
2023-03-28 05:45:33,056 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils        [] - Actor system started at <akka.tcp://flink@flink-jobmanager:6123>
2023-03-28 05:45:33,105 INFO  org.apache.flink.configuration.Configuration                 [] - Config uses fallback configuration key 'jobmanager.rpc.address' instead of key 'rest.address'
2023-03-28 05:45:33,117 INFO  org.apache.flink.runtime.blob.BlobServer                     [] - Created BLOB server storage directory /tmp/jm_ba3e9448356424b7c239a69bc9fdf63c/blobStorage
2023-03-28 05:45:33,123 INFO  org.apache.flink.runtime.blob.BlobServer                     [] - Started BLOB server at 0.0.0.0:6124 - max concurrent requests: 50 - max backlog: 1000
2023-03-28 05:45:33,128 INFO  org.apache.flink.runtime.security.token.KerberosDelegationTokenManagerFactory [] - Cannot use kerberos delegation token manager because Hadoop cannot be found in the Classpath.
2023-03-28 05:45:33,145 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          [] - No metrics reporter configured, no metrics will be exposed/reported.
2023-03-28 05:45:33,150 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils        [] - Trying to start actor system, external address flink-jobmanager:0, bind address 0.0.0.0:0.
2023-03-28 05:45:33,207 INFO  akka.event.slf4j.Slf4jLogger                                 [] - Slf4jLogger started
2023-03-28 05:45:33,212 INFO  akka.remote.RemoteActorRefProvider                           [] - Akka Cluster not in use - enabling unsafe features anyway because `akka.remote.use-unsafe-remote-features-outside-cluster` has been enabled.
2023-03-28 05:45:33,213 INFO  akka.remote.Remoting                                         [] - Starting remoting
2023-03-28 05:45:33,231 INFO  akka.remote.Remoting                                         [] - Remoting started; listening on addresses :[<akka.tcp://flink-metrics>@flink-jobmanager:43153]
2023-03-28 05:45:33,248 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils        [] - Actor system started at <akka.tcp://flink-metrics>@flink-jobmanager:43153
2023-03-28 05:45:33,276 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService             [] - Starting RPC endpoint for org.apache.flink.runtime.metrics.dump.MetricQueryService at <akka://flink-metrics/user/rpc/MetricQueryService> .
2023-03-28 05:45:33,304 INFO  org.apache.flink.runtime.dispatcher.FileExecutionGraphInfoStore [] - Initializing FileExecutionGraphInfoStore: Storage directory /tmp/executionGraphStore-b1290a8b-573a-4d2c-8a7c-2c34b494d779, expiration time 3600000, maximum cache size 52428800 bytes.
2023-03-28 05:45:33,399 INFO  org.apache.flink.configuration.Configuration                 [] - Config uses fallback configuration key 'jobmanager.rpc.address' instead of key 'rest.address'
2023-03-28 05:45:33,400 INFO  org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint   [] - Upload directory /tmp/flink-web-1db1b80d-e08a-4659-ba28-7f8f9fccd7bb/flink-web-upload does not exist. 
2023-03-28 05:45:33,401 INFO  org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint   [] - Created directory /tmp/flink-web-1db1b80d-e08a-4659-ba28-7f8f9fccd7bb/flink-web-upload for file uploads.
2023-03-28 05:45:33,407 INFO  org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint   [] - Starting rest endpoint.
2023-03-28 05:45:33,749 INFO  org.apache.flink.runtime.webmonitor.WebMonitorUtils          [] - Determined location of main cluster component log file: /opt/flink/log/flink--standalonesession-0-flink-jobmanager-564495f8d8-9kndc.log
2023-03-28 05:45:33,750 INFO  org.apache.flink.runtime.webmonitor.WebMonitorUtils          [] - Determined location of main cluster component stdout file: /opt/flink/log/flink--standalonesession-0-flink-jobmanager-564495f8d8-9kndc.out
s
can you check if you have 
plugins/s3/
path and put flink-hadoop-fs-s3.jar and flink-presto-fs-s3. jar and try running the application again ?
o
I think s3a is rather 
flink-hadoop-fs-s3
lib. presto-fs-s3 is requiring s3p path
👍 1
a
@Samrat Deb @Oleksandr Nitavskyi - Thank you for your suggestions - I tried the above and still was not able to get the filesystem working - i tried even with presto s3 plugin as suggested by @Samrat Deb However, interesting point to note - When I tried read a csv file from s3 using a simple CsvInputFormat, it was able to read s3 data just fine. So it seems like this issue is happening specifically with Delta Lake’s Flink connector only. Is there somewhere I can bring up this as an issue to be checked? Has anyone tried connecting to a delta lake stored in s3 from flink before and faced this issue?
s
you need uber jar for flink-delta-connector. check delta.io and maven for delta-flink-1.16 connector
a
So this uber jar needs to be added to lib/ folder in my flink deployment? From the doc - looks like I just need to add it as dependency in my pom.xml right? I already have this library as a maven dependency in my project. Also, where do I find the uber jar? Was not able to find it in the delta.io website
2 Views
Open in Slack
PreviousNext
Powered by