文档中心

向量数据库 Milvus 版

数据迁移

创建工作流

向量数据库 Milvus 版支持数据迁移，您可以通过工作流将其他数据源中的存量数据迁移至向量数据库 Milvus 版。本文介绍如何创建工作流。

前提条件

已创建 Milvus 实例、Flink 项目和 Flink 资源池，且 Milvus 实例与 Flink 资源池在同一私有网络中。具体操作，请参见创建实例、导入项目和创建资源池。

如数据迁移来源为 Kafka，需创建 Kafka 实例并获取接入点，且 Kafka 实例与 Milvus 实例需在同一私有网络中。具体操作，请参见创建实例和查看接入点。

如需添加 Embedding Transform 节点，Flink 项目需开启公网访问。具体操作，请参见 Flink 访问公网。

操作步骤

登录向量数据库 Milvus 版控制台。

在顶部导航栏，选择需要创建工作流的项目和地域。

在左侧导航栏，选择数据迁移，然后单击创建工作流。

在选择 Source 类型对话框中，选择数据迁移来源，然后单击去创建。

说明

当前数据迁移来源支持 Kafka、TOS、自建 Milvus 和 Viking DB。选择自建 Milvus 或 Viking DB 将跳转至数据库传输服务 DTS 创建同步任务。具体操作，请参见同步至火山引擎版 Milvus。

在创建工作流页面中，配置以下参数。

参数	子参数	说明
基本信息
工作流名称			输入工作流的名称。命名规则如下：长度为 4~128 个字符。必须以字母或数字开头和结尾。只能包含字母、数字，下划线（_）和中划线（-）。
备注（可选）			输入工作流的备注，最多支持 128 个字符。
工作流说明在工作流中配置各节点。工作流默认包含 Source 和 Destination 两个节点，分别为起点和终点。您还可以单击 Add Transform 添加最多 5 个 Transform 节点。
Source	Kafka	在类型下拉列表中选择 Kafka，然后配置以下字段。 Kafka Instance：在下拉列表中选择要迁移的 Kafka 实例。如需创建新的 Kafka 实例，请参见创建实例。 Kafka Topic：在下拉列表中选择要迁移的 Kafka 消息主题。如需创建新的 Kafka 消息主题，请参见创建 Topic。 Bootstrap Servers：输入 Kafka 实例的接入点 IP 地址。获取接入点 IP 地址的方式，请参见查看接入点。 Consumer Group：输入 Kafka 实例的消费组。获取消费组的方式，请参见查看 Group 详情。如果 Kafka 实例没有开启自动创建 Group 功能，则需要在 Kafka 实例中手动创建消费组。具体操作，请参见创建 Group。高级配置（可选）：您还可以使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 指定数据 schema 告知数据来源如何将数据解析为想要的行。 "schema": { "fields": { "id": "int", "age": "int", "name": "string", "description": "string" }, "primaryKey": { "name": "id", "columnNames": [ "id" ] } }, "kafka.config": { # Kafka 消费者客户端 ID。 "client.id": "kafka2milvus_client", # 最大拉取数据条数。 "max.poll.records": 500, # 当不存在消费 offset 时的消费位点策略。 "auto.offset.reset": "earliest", # 是否开启 offset 自动提交。 "enable.auto.commit": "true", # Kafka 接入点认证协议。获取方式请参见使用 PLAINTEXT 接入点连接实例。 "sasl.mechanism":"PLAIN" }, # 数据格式。 "format": "json", # 消费者的初始消费模式。 "start_mode": "earliest" }
Source		TOS	在类型下拉列表中选择 TOS，然后配置以下字段。 Bucket：选择 TOS 存储桶。如需创建新的 TOS 存储桶，请参见创建存储桶。 Path：选择 TOS 存储桶中的文件或者目录，选择目录时支持使用 file_filter_pattern 选项过滤文件。 Endpoint：输入 TOS 存储桶的访问域名。获取访问域名的方式，请参见查看桶详情。 Credentials Providers：凭证提供方。当前固定为 org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider。 Access Key：输入获取的访问密钥 Access Key 。获取 Access Key 的方式，请参见 Access Key（密钥）管理。 Access Secret：输入获取的访问密钥 Access Secret 。获取 Access Secret 的方式，请参见 Access Key（密钥）管理。 File Format：在下拉列表中选择文件格式。当前支持 text、csv、parquet、orc、json、excel、xml 和 binary 格式。高级配置（可选）：您还可以使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 指定数据 schema 告知数据来源如何将数据解析为想要的行。 "schema": { "fields": { "id": "int", "age": "int", "name": "string", "description": "string" }, "primaryKey": { "name": "id", "columnNames": [ "id" ] } } }
Destination			数据迁移目的地。当前固定为 Milvus，需配置以下字段。实例类型：选择 Dedicated 或 Serverless。 Milvus 实例：在下拉列表中选择要将数据迁移到的 Milvus 实例。如需创建新的 Milvus 实例，请参见创建实例。 URL ：显示 Milvus 实例的访问地址，不可修改。 Token：输入 Milvus 实例的账号和密码，用 : 分隔。 Database：输入要将数据迁移到的库名。 Collection：输入要将数据迁移到的集合名。高级配置（可选）：您还可以使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 是否开启自动 ID "enable_auto_id": true, # 批量写入大小 "batch_size": 1000, # schema 保存模式 "schema_save_mode": "RECREATE_SCHEMA", # 数据保存模式 "data_save_mode": "APPEND_DATA", # 自动创建索引，默认 false "create_index": true, # 自动加载集合，默认 false "load_collection": true }
Transform（可选）	LLM	在下拉列表中选择 LLM，然后配置以下字段。 Model Provider：在下拉列表中选择 LLM 模型提供方。 Model：输入 LLM 模型名称，例如 doubao-seed-1-6-251015。 API Key：输入用于调用 LLM 模型的 API Key。 Prompt：输入告知 LLM 模型期望结果的提示词。 Output Column：输入 LLM 模型的输出列。高级配置（可选）：您还可以使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 指定输入数据中的哪些字段值用作模型的输入，例如以下是将 name 字段值作为输入。 "inference_columns": ["name"] }
		Embedding	在下拉列表中选择 LLM，然后配置以下字段。 Model Provider：在下拉列表中选择 Embedding 模型提供方。 Model：输入 Embedding 模型名称，例如 doubao-seed-1-6-251015。 API Key：输入用于调用 Embedding 模型的 API Key。高级配置：您还可以使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 原始字段和向量字段的映射，将使用原始字段的值 Embedding 后作为向量字段的值 "vectorization_fields": { "description_vector": "description" } } 说明 Embedding 节点支持配置豆包多模态向量化模型，可对文本、图片、视频进行多个数据进行混排多模态向量化。详细信息，请参见多模态数据向量化。
		Field Mapper	在下拉列表中选择 FieldMapper，然后使用代码进行高级配置。单击字段可在右侧查看字段详细说明。配置示例如下。 { # 映射字段, 以下配置将数据中的 age 字段修改为 real_age 字段。未配置的字段将被滤不再向后传递。 "field_mapper": { "id": "id", "age": "real_age", "name": "name", "description": "description" }, }
作业模式			选择 BATCH 或 STREAMING。
Flink 资源池
私有网络			固定为当前 Milvus 实例所在的私有网络，不可修改。
Flink 项目			在下拉列表中选择所属的项目。如需创建新的项目，请参见新建项目。
Flink 资源池			在下拉列表中使用的 Flink 资源池。如需创建新的资源池，请参见创建资源池。
Flink 高级配置
并行度			输入任务全局并发数，默认值为 4。
单个 TaskManager CPU 数			在下拉列表中选择单个 TaskManager 的 CPU 核数，默认值为 1。
单个 TaskManager 内存大小			在下拉列表中选择单个 TaskManager 的内存大小，选项将根据您设置的 CPU 核数自动调整。
单个 TaskManager slot 数			单个 TaskManager 的 Slot 数量，默认值为 4。
JobManager CPU 数			JobManager 的 CPU 核数，默认值为 1。
JobManager 内存大小			JobManager 的内存大小，选项将根据您设置的 CPU 核数自动调整。

根据需要单击仅创建或创建并启动。

仅创建：仅创建工作流，不实际运行。创建后，系统会返回数据迁移页面，工作流显示为初始化状态。可在工作流列表中，单击目标工作流操作列下的启动来启动数据迁移工作流。

创建并启动：完成创建并开始运行数据迁移工作流。

最近更新时间：2026.08.27 17:54:25

这个页面对您有帮助吗？

有用

无用