创建数据管道（Java SDK V2） - 对象存储 OSS

通过 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);
        }
    }
}

对象存储 OSS：创建数据管道（Java SDK V2）

前提条件

示例代码

相关文档