SB5.1 - OCI - Deploy/KnowledgePlatform/FlinkJobs - S3 storage #327

ddevadat · 2023-01-01T14:11:17Z

ddevadat
Jan 1, 2023

CSP: OCI

Could you please share the steps to configure Deploy/KnowledgePlatform/FlinkJobs to use s3 as a storage option.

I tried to use s3 compliant api, but jobmanager is failing to start

2023-01-01 13:48:31,225 DEBUG org.apache.flink.fs.s3.common.AbstractS3FileSystemFactory [] - Loading Hadoop configuration for Hadoop s3a file system 2023-01-01 13:48:31,263 DEBUG org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess [] - Initialization of the JobMasterService for job 58d3261e72c02a4f34ee1824d8a98de2 under leader id 00000000-0000-0000-0000-000000000000 failed. org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster. at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source) [?:?] at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source) [?:?] at java.util.concurrent.CompletableFuture.postComplete(Unknown Source) [?:?] at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) [?:?] at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) [?:?] at java.util.concurrent.FutureTask.run(Unknown Source) [?:?] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) [?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?] at java.lang.Thread.run(Unknown Source) [?:?] Caused by: java.util.concurrent.CompletionException: java.lang.NoSuchMethodError: 'void org.apache.hadoop.conf.Configuration.reloadExistingConfigurations()' at java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) ~[?:?] at java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source) ~[?:?] ... 7 more Caused by: java.lang.NoSuchMethodError: 'void org.apache.hadoop.conf.Configuration.reloadExistingConfigurations()' at org.apache.hadoop.fs.s3a.S3AFileSystem.addDeprecatedKeys(S3AFileSystem.java:218) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5] at org.apache.hadoop.fs.s3a.S3AFileSystem.<clinit>(S3AFileSystem.java:222) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5] at org.apache.flink.fs.s3hadoop.S3FileSystemFactory.createHadoopFileSystem(S3FileSystemFactory.java:71) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5] at org.apache.flink.fs.s3.common.AbstractS3FileSystemFactory.create(AbstractS3FileSystemFactory.java:122) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5] at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:506) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:407) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.core.fs.Path.getFileSystem(Path.java:274) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess.<init>(FsCheckpointStorageAccess.java:64) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.state.filesystem.FsStateBackend.createCheckpointStorage(FsStateBackend.java:527) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.<init>(CheckpointCoordinator.java:322) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.<init>(CheckpointCoordinator.java:241) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.enableCheckpointing(DefaultExecutionGraph.java:448) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:311) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:107) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:342) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:190) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:122) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:132) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:110) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:340) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:317) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:107) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) ~[flink-dist_2.12-1.13.5.jar:1.13.5] at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) ~[flink-dist_2.12-1.13.5.jar:1.13.5] ... 7 more

Answered by ddevadat

Jan 11, 2023

Eventually solved this by using flink-s3-fs-presto

For this added the dependency while building the cloud-storage-sdk package

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-s3-fs-presto</artifactId>
    <version>1.13.5</version>
</dependency>

Then in the flink helm chart deployment...added the presto variable for passing s3 credentials like below

        args: ["start-foreground",
               "--job-classname={{ .Values.job_classname }}",
               {{- if eq .Values.csp "oci" }}
               "-Dpresto.s3.access-key={{ .Values.s3_access_key}}",
               "-Dpresto.s3.secret-key={{ .Values.s3_secret_key }}",
               "-Dpresto.s3.endpoint=…

View full answer

ddevadat · 2023-01-02T06:32:30Z

ddevadat
Jan 2, 2023
Author

any update on this

0 replies

ddevadat · 2023-01-03T01:30:35Z

ddevadat
Jan 3, 2023
Author

any update on this

0 replies

beepdot · 2023-01-03T05:51:52Z

beepdot
Jan 3, 2023

@AmiableAnil @reshmi-nair Can you please provide your inputs on this thread?

0 replies

ddevadat · 2023-01-10T05:16:18Z

ddevadat
Jan 10, 2023
Author

i builded the cloudstorage sdk for using OCI object storage

https://gist.github.com/ddevadat/7a3938833b2bf2908c6b0773bca04e93

the key things things changed
guava
jcloud_all
hadoop-azure,hadoop-common,hadoop-aws

i tried to start kp flink jobs, questionset-publish is working fine and its creating checkpoint directories in oci object storage bucket

However other jobmanagers are failing with the below error

`

java.nio.file.AccessDeniedException: flink-check-points-store: org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException: No AWS Credentials provided by SimpleAWSCredentialsProvider EnvironmentVariableCredentialsProvider InstanceProfileCredentialsProvider : com.amazonaws.SdkClientException: The requested metadata is not found at http://169.254.169.254/latest/meta-data/iam/security-credentials/
at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:187) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.Invoker.once(Invoker.java:111) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.Invoker.lambda$retry$3(Invoker.java:265) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:322) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.Invoker.retry(Invoker.java:261) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.Invoker.retry(Invoker.java:236) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.S3AFileSystem.verifyBucketExists(S3AFileSystem.java:391) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:322) ~[asset-enrichment-1.0.0.jar:?]
at org.apache.flink.fs.s3.common.AbstractS3FileSystemFactory.create(AbstractS3FileSystemFactory.java:123) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5]
at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:506) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:407) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.core.fs.Path.getFileSystem(Path.java:274) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess.(FsCheckpointStorageAccess.java:64) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.state.filesystem.FsStateBackend.createCheckpointStorage(FsStateBackend.java:527) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.(StreamTask.java:341) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.(StreamTask.java:308) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.(SourceStreamTask.java:76) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.(SourceStreamTask.java:72) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:?]
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source) ~[?:?]
at jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source) ~[?:?]
at java.lang.reflect.Constructor.newInstance(Unknown Source) ~[?:?]
at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1529) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:730) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException: No AWS Credentials provided by SimpleAWSCredentialsProvider EnvironmentVariableCredentialsProvider InstanceProfileCredentialsProvider : com.amazonaws.SdkClientException: The requested metadata is not found at http://169.254.169.254/latest/meta-data/iam/security-credentials/
`

0 replies

ddevadat · 2023-01-11T04:07:04Z

ddevadat
Jan 11, 2023
Author

when using hadoop-aws

For using path style access hadoop-aws has to be 2.8.0 or higher
path style access introduced in hadoop-2.8.0
https://issues.apache.org/jira/browse/HADOOP-12963

2)After this its able to create checkpoint buckets, but not able to interact with the bucket becuase it checks for if the bucket exists. This call
used v2 signing method, which will fail for s3 compliant storage

For this we need to use hadoop 3.3.1 or higher , because hadoop introduced this feature to disable the bucket verification check
bucket probe disabled
https://issues.apache.org/jira/browse/HADOOP-17454

So now we know that minimum hadoop version required is 3.3.1

But on using 3.3.1, we get errors like

Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
    at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:316) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
    at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
    at java.util.concurrent.CompletableFuture$UniApply.tryFire(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture$Completion.run(Unknown Source) ~[?:?]
    at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
    at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
    at java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture.postComplete(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) ~[?:?]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) ~[?:?]
    at java.util.concurrent.FutureTask.run(Unknown Source) ~[?:?]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
    at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: java.util.concurrent.CompletionException: java.lang.NoSuchMethodError: 'void org.apache.hadoop.fs.s3a.WriteOperationHelper.<init>(org.apache.hadoop.fs.s3a.S3AFileSystem, org.apache.hadoop.conf.Configuration)'
    at java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source) ~[?:?]
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) ~[?:?]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) ~[?:?]
    at java.util.concurrent.FutureTask.run(Unknown Source) ~[?:?]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
    at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: java.lang.NoSuchMethodError: 'void org.apache.hadoop.fs.s3a.WriteOperationHelper.<init>(org.apache.hadoop.fs.s3a.S3AFileSystem, org.apache.hadoop.conf.Configuration)'
    at org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper$InternalWriteOperationHelper.<init>(HadoopS3AccessHelper.java:148) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5]
    at org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.<init>(HadoopS3AccessHelper.java:56) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5]
    at org.apache.flink.fs.s3hadoop.S3FileSystemFactory.getS3AccessHelper(S3FileSystemFactory.java:97) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5]
    at org.apache.flink.fs.s3.common.AbstractS3FileSystemFactory.create(AbstractS3FileSystemFactory.java:149) ~[flink-s3-fs-hadoop-1.13.5.jar:1.13.5]
    at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:506) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
    at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:407) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
    at org.apache.flink.core.fs.Path.getFileSystem(Path.java:274) ~[flink-dist_2.12-1.13.5.jar:1.13.5]

1 reply

gandham-santhosh Jan 11, 2023

@maheshkumargangula @anandp504 @reshmi-nair any thoughts on this?

ddevadat · 2023-01-11T12:05:15Z

ddevadat
Jan 11, 2023
Author

Eventually solved this by using flink-s3-fs-presto

For this added the dependency while building the cloud-storage-sdk package

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-s3-fs-presto</artifactId>
    <version>1.13.5</version>
</dependency>

Then in the flink helm chart deployment...added the presto variable for passing s3 credentials like below

        args: ["start-foreground",
               "--job-classname={{ .Values.job_classname }}",
               {{- if eq .Values.csp "oci" }}
               "-Dpresto.s3.access-key={{ .Values.s3_access_key}}",
               "-Dpresto.s3.secret-key={{ .Values.s3_secret_key }}",
               "-Dpresto.s3.endpoint={{ .Values.s3_endpoint }}",
               "-Dpresto.s3.region={{ .Values.s3_region }}",
               "-Dpresto.s3.path-style-access={{ .Values.s3_path_style_access }}",

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbird-ED (Sunbird for Education)

SB5.1 - OCI - Deploy/KnowledgePlatform/FlinkJobs - S3 storage #327

{{title}}

Replies: 6 comments 1 reply

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Sunbird-ED (Sunbird for Education)

SB5.1 - OCI - Deploy/KnowledgePlatform/FlinkJobs - S3 storage #327

ddevadat Jan 1, 2023

Replies: 6 comments · 1 reply

ddevadat Jan 2, 2023 Author

ddevadat Jan 3, 2023 Author

beepdot Jan 3, 2023

ddevadat Jan 10, 2023 Author

ddevadat Jan 11, 2023 Author

gandham-santhosh Jan 11, 2023

ddevadat Jan 11, 2023 Author

ddevadat
Jan 1, 2023

Replies: 6 comments 1 reply

ddevadat
Jan 2, 2023
Author

ddevadat
Jan 3, 2023
Author

beepdot
Jan 3, 2023

ddevadat
Jan 10, 2023
Author

ddevadat
Jan 11, 2023
Author

ddevadat
Jan 11, 2023
Author