SB5.1 - OCI - Build/DataPipeline/AnalyticsCore and cloud-store-sdk #376

ddevadat · 2023-01-31T00:45:28Z

ddevadat
Jan 31, 2023

We are trying to add oci oss support to Analytics core.

https://github.com/project-sunbird/sunbird-analytics-core.git
release-5.0.0_RC2

We are using cloud-store-sdk: 1.4.6
Need to know a few things.

For hadoop configuration , for S3, its using s3n filesystem.

  def setS3Conf(sc: SparkContext) = {
    JobLogger.log("Configuring S3 AccessKey& SecrateKey to SparkContext")
    sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", AppConf.getAwsKey());
    sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", AppConf.getAwsSecret());

    val storageEndpoint = AppConf.getConfig("cloud_storage_endpoint")
    if (!"".equalsIgnoreCase(storageEndpoint)) {
      sc.hadoopConfiguration.set("fs.s3n.endpoint", storageEndpoint)
    }
  }

For s3n,
https://hadoop.apache.org/docs/r2.7.2/hadoop-aws/tools/hadoop-aws/index.html

Didnt find a parameter for setting endpoint like fs.s3n.endpoint

anandp504 · 2023-01-31T10:00:18Z

anandp504
Jan 31, 2023

@ddevadat I understand that the OCI cloud storage is s3 compatible. If we set the fs.s3n.awsAccessKeyId and fs.s3n.awsSecretAccessKey, it should work with OCI cloud storage. The storage endpoint is not mandatory.

0 replies

ddevadat · 2023-02-01T00:20:10Z

ddevadat
Feb 1, 2023
Author

isnt s3n client gone obsolete. The documentation says to upgrade to s3a as s3n is no longer maintained.

4 replies

Krishnaj20 Feb 9, 2023
Maintainer

@anandp504 - can you check and respond?

Krishnaj20 Feb 14, 2023
Maintainer

@anandp504 @maheshkumargangula - can you check and respond to this query please?

@vinukumar-vs

ddevadat Apr 18, 2023
Author

@anandp504
This is not working with s3n . I have set the context to include only
sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", AppConf.getAwsKey());
sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", AppConf.getAwsSecret());

jets3t.properties is also updated. But getting permission denied


23/04/18 03:34:24 INFO DAGScheduler: Job 11 finished: pivot at ProgressExhaustJob.scala:72, took 6.512862 s
org.apache.hadoop.security.AccessControlException: Permission denied: s3n://bucket/progress-exhaust/73345E4BBC80BA02AB83B870E0301E75/0137588915168542720_progress_20230418/_tmp
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.processException(Jets3tNativeFileSystemStore.java:449)
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.processException(Jets3tNativeFileSystemStore.java:427)
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.handleException(Jets3tNativeFileSystemStore.java:411)
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:181)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:191)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
	at org.apache.hadoop.fs.s3native.$Proxy48.retrieveMetadata(Unknown Source)
	at org.apache.hadoop.fs.s3native.NativeS3FileSystem.getFileStatus(NativeS3FileSystem.java:476)
	at org.apache.hadoop.fs.s3native.NativeS3FileSystem.delete(NativeS3FileSystem.java:420)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.$anonfun$delete$1(HadoopFileUtil.scala:36)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.$anonfun$delete$1$adapted(HadoopFileUtil.scala:34)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
	at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
	at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:38)
	at scala.collection.TraversableLike.map(TraversableLike.scala:238)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
	at scala.collection.AbstractTraversable.map(Traversable.scala:108)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.delete(HadoopFileUtil.scala:34)
	at org.ekstep.analytics.framework.util.DatasetExt.saveToBlobStore(DatasetUtil.scala:76)
	at org.sunbird.analytics.exhaust.collection.BaseCollectionExhaustJob.$anonfun$processBatches$8(BaseCollectionExhaustJob.scala:354)
	at scala.collection.immutable.List.map(List.scala:286)
	at org.sunbird.analytics.exhaust.collection.BaseCollectionExhaustJob.$anonfun$processBatches$6(BaseCollect

Krishnaj20 Apr 25, 2023
Maintainer

@anandp504 - Is this an open issue which has to be added to Observ's backlog? Or is this resolved as part of the weekend migration? Pls confirm.

ddevadat · 2023-04-25T21:43:53Z

ddevadat
Apr 25, 2023
Author

@Krishnaj20 @anandp504
observation is s3n is not at all stable. The analytics job keeps failing at downloading the files

23/04/25 21:38:43 INFO NativeS3FileSystem: OutputStream for key 'ml_reports/ml-project-status-exhaust/458FE02DF03C2E21700AC730BB2CDF0D_20230425.csv' closed. Now beginning upload
23/04/25 21:38:43 WARN RestStorageService: Could not reset the buffer
java.io.IOException: Resetting to invalid mark
	at java.base/java.io.BufferedInputStream.reset(BufferedInputStream.java:454)
	at org.jets3t.service.impl.rest.httpclient.RestStorageService.putObjectImpl(RestStorageService.java:1901)
	at org.jets3t.service.StorageService.putObject(StorageService.java:840)
	at org.jets3t.service.S3Service.putObject(S3Service.java:2122)
	at org.jets3t.service.S3Service.putObject(S3Service.java:2264)
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.storeFile(Jets3tNativeFileSystemStore.java:122)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:191)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
	at org.apache.hadoop.fs.s3native.$Proxy40.storeFile(Unknown Source)
	at org.apache.hadoop.fs.s3native.NativeS3FileSystem$NativeS3FsOutputStream.close(NativeS3FileSystem.java:284)
	at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:72)
	at org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:106)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.copyMerge(HadoopFileUtil.scala:65)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.copyMerge(HadoopFileUtil.scala:48)
	at org.ekstep.analytics.framework.util.DatasetExt.saveToBlobStore(DatasetUtil.scala:97)
	at org.ekstep.analytics.util.BaseDruidQueryProcessor.saveReport(BaseDruidQueryProcessor.scala:172)
	at org.ekstep.analytics.util.BaseDruidQueryProcessor.saveReport$(BaseDruidQueryProcessor.scala:134)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.saveReport(OnDemandDruidExhaustJob.scala:30)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.$anonfun$druidPostProcess$5(OnDemandDruidExhaustJob.scala:114)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.$anonfun$druidPostProcess$5$adapted(OnDemandDruidExhaustJob.scala:98)
	at scala.collection.immutable.List.foreach(List.scala:392)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.druidPostProcess(OnDemandDruidExhaustJob.scala:98)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.$anonfun$processRequest$1(OnDemandDruidExhaustJob.scala:137)
	at org.ekstep.analytics.framework.util.CommonUtil$.time(CommonUtil.scala:548)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.processRequest(OnDemandDruidExhaustJob.scala:137)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.$anonfun$execute$1(OnDemandDruidExhaustJob.scala:203)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
	at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
	at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)
	at scala.collection.TraversableLike.map(TraversableLike.scala:238)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
	at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:198)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.execute(OnDemandDruidExhaustJob.scala:167)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.$anonfun$main$1(OnDemandDruidExhaustJob.scala:50)
	at org.ekstep.analytics.framework.util.CommonUtil$.time(CommonUtil.scala:548)
	at org.ekstep.analytics.exhaust.OnDemandDruidExhaustJob$.main(OnDemandDruidExhaustJob.scala:50)
	at org.ekstep.analytics.job.JobExecutor$.main(JobExecutor.scala:16)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at optional.Application.callWithOptions(application.scala:251)
	at optional.Application.callWithOptions$(application.scala:223)
	at org.ekstep.analytics.job.JobExecutor$.callWithOptions(JobExecutor.scala:8)
	at optional.Application.main(application.scala:258)
	at optional.Application.main$(application.scala:255)
	at org.ekstep.analytics.job.JobExecutor$.main(JobExecutor.scala:8)
	at org.ekstep.analytics.job.JobExecutor.main(JobExecutor.scala)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
	at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:951)
	at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
	at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
	at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
	at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1039)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1048)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

We tried tuning the jets3t.properties..but none is working to resolve this reset mark error

storage-service.request-signature-version=AWS4-HMAC-SHA256

s3service.s3-endpoint=XXXXXXX
s3service.disable-dns-buckets=true
s3service.https-only=true
s3service.default-bucket-location=XXXXXX
uploads.stream-retry-buffer-size=131073

This is severely affecting the report generation in production .

2 replies

Krishnaj20 Apr 26, 2023
Maintainer

@gandham-santhosh - who can help resolve this issue?

@anandp504 @SanthoshVasabhaktula @vinukumar-vs @maheshkumargangula

SanthoshVasabhaktula Apr 26, 2023
Collaborator

@ddevadat - This isn't an issue with the code. The same code works with AWS. Please check whether s3n API is compatible with OCI or not. If not, do an upgrade to s3a and if it works do send a PR back. In addition, did this work in QA before directly testing in production?

@Krishnaj20 - Please tag Reshmi or Kumar for any assistance to @ddevadat in resolving this issue at OCI.

ddevadat · 2023-04-26T06:07:13Z

ddevadat
Apr 26, 2023
Author

the file is actually getting uploaded in the bucket
The below warning is still there, but as the file is getting uploaded , i think we can ignore this. This coming from the jets3t lib.

23/04/25 21:38:43 INFO NativeS3FileSystem: OutputStream for key 'ml_reports/ml-project-status-exhaust/458FE02DF03C2E21700AC730BB2CDF0D_20230425.csv' closed. Now beginning upload
23/04/25 21:38:43 WARN RestStorageService: Could not reset the buffer
java.io.IOException: Resetting to invalid mark

Please note that we have used jets3t version 0.97 instead of 0.94

0.94 version of jets3t doesnt work with s3 compliant storage.

it throws errors like

java.lang.ClassCastException: class java.lang.String cannot be cast to class java.util.Date (java.lang.String and java.util.Date are in module java.base of loader 'bootstrap')
	at org.jets3t.service.model.StorageObject.getLastModifiedDate(StorageObject.java:376)
	at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:176)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:191)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
	at org.apache.hadoop.fs.s3native.$Proxy40.retrieveMetadata(Unknown Source)
	at org.apache.hadoop.fs.s3native.NativeS3FileSystem.getFileStatus(NativeS3FileSystem.java:483)
	at org.apache.hadoop.fs.s3native.NativeS3FileSystem.delete(NativeS3FileSystem.java:420)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.$anonfun$delete$1(HadoopFileUtil.scala:36)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.$anonfun$delete$1$adapted(HadoopFileUtil.scala:34)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
	at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
	at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:38)
	at scala.collection.TraversableLike.map(TraversableLike.scala:238)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
	at scala.collection.AbstractTraversable.map(Traversable.scala:108)
	at org.ekstep.analytics.framework.util.HadoopFileUtil.delete(HadoopFileUtil.scala:34)
	at org.ekstep.analytics.framework.util.DatasetExt.saveToBlobStore(DatasetUtil.scala:76)

1 reply

SanthoshVasabhaktula Apr 26, 2023
Collaborator

@ddevadat - can you send a PR for the version upgrade?

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbird-ED (Sunbird for Education)

SB5.1 - OCI - Build/DataPipeline/AnalyticsCore and cloud-store-sdk #376

{{title}}

Replies: 4 comments 7 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

Select a reply

Sunbird-ED (Sunbird for Education)

SB5.1 - OCI - Build/DataPipeline/AnalyticsCore and cloud-store-sdk #376

ddevadat Jan 31, 2023

Replies: 4 comments · 7 replies

anandp504 Jan 31, 2023

ddevadat Feb 1, 2023 Author

Krishnaj20 Feb 9, 2023 Maintainer

Krishnaj20 Feb 14, 2023 Maintainer

ddevadat Apr 18, 2023 Author

Krishnaj20 Apr 25, 2023 Maintainer

ddevadat Apr 25, 2023 Author

Krishnaj20 Apr 26, 2023 Maintainer

SanthoshVasabhaktula Apr 26, 2023 Collaborator

ddevadat Apr 26, 2023 Author

SanthoshVasabhaktula Apr 26, 2023 Collaborator

ddevadat
Jan 31, 2023

Replies: 4 comments 7 replies

anandp504
Jan 31, 2023

ddevadat
Feb 1, 2023
Author

Krishnaj20 Feb 9, 2023
Maintainer

Krishnaj20 Feb 14, 2023
Maintainer

ddevadat Apr 18, 2023
Author

Krishnaj20 Apr 25, 2023
Maintainer

ddevadat
Apr 25, 2023
Author

Krishnaj20 Apr 26, 2023
Maintainer

SanthoshVasabhaktula Apr 26, 2023
Collaborator

ddevadat
Apr 26, 2023
Author

SanthoshVasabhaktula Apr 26, 2023
Collaborator