建立資料管道（Java SDK V2） - Object Storage Service

通過 OSS Java SDK V2 調用 PutDataPipelineConfiguration 建立一條資料管道（DataPipeline）規則。規則建立後，OSS 自動調用您的阿里雲百鍊對源 Bucket 中的未經處理資料進行向量化，並將向量資料寫入向量 Bucket 的指定 Index。

前提條件

已安裝 OSS Java SDK V2。使用 Maven 時，請在 pom.xml 中添加依賴 com.aliyun:alibabacloud-oss-v2。
已通過環境變數配置訪問憑證。本文樣本使用 EnvironmentVariableCredentialsProvider 從環境變數 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET 讀取憑證。
資料管道（DataPipeline）相關介面由 OSS 透傳至底層工作流程服務，請確保源通用 Bucket、目標向量 Bucket 位於同一地區。

範例程式碼

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;
import java.util.Arrays;
import java.util.Collections;

public class PutDataPipelineConfigurationSample {
    public static void main(String[] args) throws Exception {
        String role = "acs:ram::<AccountId>:role/my-data-pipeline-role";
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            DataPipelineSourceFilterConfiguration filterConfig =
                    DataPipelineSourceFilterConfiguration.newBuilder()
                            .prefixSet(Arrays.asList("data/"))
                            .objectMediaTypes(Arrays.asList("image", "video"))
                            .build();

            DataPipelineSource source = DataPipelineSource.newBuilder()
                    .inputBucket("my-bucket")
                    .inputDataScope("All")
                    .ignoreDelete(true)
                    .filterConfiguration(filterConfig)
                    .build();

            DataPipelineEmbeddingConfiguration embeddingConfig =
                    DataPipelineEmbeddingConfiguration.newBuilder()
                            .embeddingProvider("bailian")
                            .apiKey("sk-xxxx")
                            .model("qwen2.5-vl-embedding")
                            .fps(1.0f)
                            .build();

            DataPipelineDestination destination = DataPipelineDestination.newBuilder()
                    .vectorBucketName("my-vector-bucket")
                    .vectorIndexNames(Collections.singletonList("my-index"))
                    .build();

            PutDataPipelineConfigurationConfiguration config =
                    PutDataPipelineConfigurationConfiguration.newBuilder()
                            .dataPipelineDescription("Data pipeline sample")
                            .sources(Collections.singletonList(source))
                            .dataPipelineEmbeddingConfiguration(embeddingConfig)
                            .destination(destination)
                            .build();

            PutDataPipelineConfigurationResult result = client.putDataPipelineConfiguration(
                    PutDataPipelineConfigurationRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .role(role)
                            .putDataPipelineConfigurationConfiguration(config)
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

Object Storage Service：建立資料管道（Java SDK V2）

前提條件

範例程式碼

相關文檔