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

通过 OSS Go SDK V2 调用 PutDataPipelineConfiguration 创建一条数据管道（DataPipeline）规则。规则创建后，OSS 自动调用您的阿里云百炼对源 Bucket 中的原始数据进行向量化，并写入向量 Bucket 的指定 Index。

前提条件

已安装 OSS Go SDK V2，可通过 go get github.com/aliyun/alibabacloud-oss-go-sdk-v2 获取。
已通过环境变量配置访问凭证。本文示例使用 credentials.NewEnvironmentVariableCredentialsProvider() 从环境变量 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET 读取凭证。

示例代码

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	role := "acs:ram::<AccountId>:role/my-data-pipeline-role"
	apiKey := "sk-xxxx"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.PutDataPipelineConfiguration(context.TODO(), &dataprocess.PutDataPipelineConfigurationRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
		Role:             oss.Ptr(role),
		DataPipelineConfiguration: &dataprocess.DataPipelineConfiguration{
			DataPipelineDescription: oss.Ptr("Vectorize business data using the BERT multimodal model"),
			Sources: []dataprocess.DataPipelineSource{
				{
					InputBucket:    oss.Ptr("bucket"),
					InputDataScope: oss.Ptr("All"),
					FilterConfiguration: &dataprocess.DataPipelineSourceFilterConfiguration{
						PrefixSet:        []string{"prefix1"},
						ObjectMediaTypes: []string{"text"},
					},
				},
			},
			DataPipelineEmbeddingConfiguration: &dataprocess.DataPipelineEmbeddingConfiguration{
				ApiKey:            oss.Ptr(apiKey),
				EmbeddingProvider: oss.Ptr("bailian"),
				FPS:               oss.Ptr(float64(1)),
				Model:             oss.Ptr("qwen2.5-vl-embedding"),
			},
			Destination: &dataprocess.DataPipelineDestination{
				VectorBucketName:    oss.Ptr("my-vector-bucket"),
				VectorIndexNames:    []string{"index"},
				VectorKeyPrefix:     oss.Ptr("prefix"),
				ObjectTagToMetadata: []string{"key1"},
				UsermetaToMetadata:  []string{"x-oss-meta-key1"},
			},
			DataPipelineError: &dataprocess.DataPipelineError{
				ErrorMode:   oss.Ptr("ignoreAndRecord"),
				ErrorBucket: oss.Ptr("my-error-bucket"),
				ErrorPrefix: oss.Ptr("error-output/"),
			},
		},
	})

	if err != nil {
		log.Fatalf("failed to put pipeline configuration %v", err)
	}
	log.Printf("put pipeline configuration result:%#v\n", result)
}

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

前提条件

示例代码

相关文档