forked from raystack/dagger
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Dart Support for OSS Service Provider (#45)
* Add gradle tasks to minimal and dependencies to maven local
* Add capability to dagger to read python udfs from Ali(oss) and Tencent(cosn) storage services
Given the configuration provided correctly. Set the below environment
variables accordingly to access the files stored in the respective
bucket.
Ali(oss)
- OSS_ACCESS_KEY_ID
- OSS_ACCESS_KEY_SECRET
Tencent(cos)
- COS_SECRET_ID
- COS_SECRET_KEY
- COS_REGION
* OSS client endpoint should be configurable via ENV variable
* COS filesystem high availability support
If you need to use COS filesystem for the dagger, provide the cos
bucket/key configuration in the state.backend.fs.checkpointdir,
state.savepoints.dir, high-availability.storageDir to flinkdeployment
manifest.
If the filesystem protocol begins with cosn for the above
configurations, dagger uses the below configurations provided in the
flinkdeployment manifest file.
fs.cosn.impl: org.apache.hadoop.fs.CosFileSystem
fs.AbstractFileSystem.cosn.impl: org.apache.hadoop.fs.CosN
fs.cosn.userinfo.secretId: <secretID>
fs.cosn.userinfo.secretKey: <secretKey>
fs.cosn.bucket.region: <region>
fs.cosn.bucket.endpoint_suffix: <tencent-provided-prefix.xyz.com>
* Fix checkstyle and made constants as static variables
* Refactor Dart Feature to plug other object storage service providers
* test checkstyle fix
* Dart Support for OSS Service Provider
* fix checkstyle
---------
Co-authored-by: Raju G T <[email protected]>
Co-authored-by: rajuGT <[email protected]>- Loading branch information
3 people
authored
Jan 23, 2025
1 parent
dc18103
commit fd991c4
Showing
13 changed files
with
238 additions
and
57 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
22 changes: 22 additions & 0 deletions
22
...ain/java/com/gotocompany/dagger/functions/udfs/scalar/dart/store/DartDataStoreClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| package com.gotocompany.dagger.functions.udfs.scalar.dart.store; | ||
|
|
||
| import com.gotocompany.dagger.common.metrics.managers.GaugeStatsManager; | ||
| import com.gotocompany.dagger.functions.exceptions.BucketDoesNotExistException; | ||
| import com.gotocompany.dagger.functions.exceptions.TagDoesNotExistException; | ||
|
|
||
| public interface DartDataStoreClient { | ||
|
|
||
| /** | ||
| * If a client could provide implementation to this, use the default data store, else implement DartDataStore along with client implementation. | ||
| * | ||
| * @param udfName either "DartGet" or "DartContains" | ||
| * @param gaugeStatsManager an instrumentation provider | ||
| * @param bucketName name of the object storage service bucket | ||
| * @param dartName from the bucket, this would be either dart-get/path/to/file.json or dart-contains/path/to/file.json | ||
| * @return Content of the file in String format from abc://bucket-name/dart-(get/contains)/path/to/file.json | ||
| * @throws TagDoesNotExistException if tag doesn't exist | ||
| * @throws BucketDoesNotExistException if bucket doesn't exist | ||
| */ | ||
| String fetchJsonData(String udfName, GaugeStatsManager gaugeStatsManager, String bucketName, String dartName) | ||
| throws TagDoesNotExistException, BucketDoesNotExistException; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
68 changes: 68 additions & 0 deletions
68
.../main/java/com/gotocompany/dagger/functions/udfs/scalar/dart/store/oss/OssDartClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package com.gotocompany.dagger.functions.udfs.scalar.dart.store.oss; | ||
|
|
||
| import com.aliyun.core.utils.IOUtils; | ||
| import com.aliyun.oss.OSS; | ||
| import com.aliyun.oss.OSSClientBuilder; | ||
| import com.aliyun.oss.common.auth.CredentialsProviderFactory; | ||
| import com.aliyun.oss.model.OSSObject; | ||
| import com.aliyuncs.exceptions.ClientException; | ||
| import com.gotocompany.dagger.common.metrics.managers.GaugeStatsManager; | ||
| import com.gotocompany.dagger.functions.exceptions.TagDoesNotExistException; | ||
| import com.gotocompany.dagger.functions.udfs.scalar.dart.DartAspects; | ||
| import com.gotocompany.dagger.functions.udfs.scalar.dart.store.DartDataStoreClient; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
|
|
||
| import static com.gotocompany.dagger.common.core.Constants.UDF_TELEMETRY_GROUP_KEY; | ||
|
|
||
| public class OssDartClient implements DartDataStoreClient { | ||
|
|
||
| private static final String ENV_OSS_ENDPOINT = "OSS_ENDPOINT"; | ||
| private static final String DEFAULT_OSS_ENDPOINT = "oss-ap-southeast-1.aliyuncs.com"; | ||
|
|
||
| private static final Double BYTES_TO_KB = 1024.0; | ||
| private static final String DART_PATH = "dartpath"; | ||
|
|
||
| private final OSS libOssClient; | ||
|
|
||
| /** | ||
| * Instantiates a new Oss client. | ||
| */ | ||
| public OssDartClient() { | ||
| String endpoint = System.getenv(ENV_OSS_ENDPOINT); | ||
| if (endpoint == null || endpoint.isEmpty()) { | ||
| endpoint = DEFAULT_OSS_ENDPOINT; | ||
| } | ||
| try { | ||
| libOssClient = new OSSClientBuilder().build(endpoint, CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider()); | ||
| } catch (ClientException e) { | ||
| throw new RuntimeException("failed to initialise oss client", e); | ||
| } | ||
| } | ||
|
|
||
| public String fetchJsonData(String udfName, GaugeStatsManager gaugeStatsManager, String bucketName, String dartName) { | ||
| OSSObject ossObject = libOssClient.getObject(bucketName, dartName); | ||
| String dartJson; | ||
| byte[] contentByteArray; | ||
| try (InputStream inputStream = ossObject.getObjectContent()) { | ||
| contentByteArray = IOUtils.toByteArray(inputStream); | ||
| dartJson = new String(contentByteArray); | ||
| } catch (IOException e) { | ||
| throw new TagDoesNotExistException("Could not find the content in oss for + dartName", e); | ||
| } | ||
| gaugeStatsManager.registerString(UDF_TELEMETRY_GROUP_KEY, udfName, DartAspects.DART_GCS_PATH.getValue(), dartName); | ||
| gaugeStatsManager.registerDouble(DART_PATH, dartName, DartAspects.DART_GCS_FILE_SIZE.getValue(), contentByteArray.length / BYTES_TO_KB); | ||
| return dartJson; | ||
| } | ||
|
|
||
| /** | ||
| * Instantiates a new OSS client. | ||
| * This constructor used for unit test purposes. | ||
| * | ||
| * @param libOssClient the storage | ||
| */ | ||
| public OssDartClient(OSS libOssClient) { | ||
| this.libOssClient = libOssClient; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.