オフライン同期による埋め込みベクトル化 - DataWorks - Alibaba Cloud ドキュメントセンター

Alibaba Cloud DataWorks の Data Integration における埋め込みベクトル化機能を使用すると、OSS、MaxCompute、HDFS などの異種ソースからデータを抽出し、ベクトルに変換して、Milvus、Elasticsearch、OpenSearch などのベクトルデータベース、または Hologres のベクトルテーブルなどの送信先に書き込むことができます。この機能により、ETL が効率化され、ナレッジのベクトル化が簡素化され、RAG などの AI シナリオが可能になります。

埋め込みを使用する理由

大規模言語モデル (LLM) 技術が進化し続ける中で、ビジネス価値を創造するためには、プライベートナレッジをこれらのモデルに統合することが不可欠です。Retrieval-Augmented Generation (RAG) は、これを実現するための主要なアプローチです。データをベクトル表現にエンコードし、ベクトルデータベースを使用して効率的な取得を行うことで、RAG は LLM に正確で信頼性が高く、動的に更新されるドメインナレッジを提供します。

ビジネスデータは、Object Storage Service (OSS)、MaxCompute、HDFS、MySQL、Oracle、メッセージキューなどの異種データソースに散在している可能性があります。このデータを埋め込み、結果のベクトルを Milvus、OpenSearch、Elasticsearch などのベクトルデータベース、または Hologres のベクトルテーブルなどのベクトル対応のターゲットに書き込む必要があります。このプロセスでは、複雑な ETL スクリプトを作成し、さまざまな種類のデータソースに適応させる必要があります。データパイプラインの複数のステージ (抽出、変換、埋め込み、書き込み) は、長く、密結合したプロセスを生み出し、モデルの反復を大幅に遅らせます。

特徴

DataWorks の Data Integration における embedding 機能を使用すると、単一のデータパイプラインでデータの抽出、埋め込みの生成、および ベクトルデータベース への書き込みが可能です。この エンドツーエンド の自動化により、開発の複雑さが軽減され、ナレッジ更新のレイテンシが短縮され、RAG、インテリジェントカスタマーサービス、検索とレコメンデーション などのシナリオで効率的な ナレッジインジェスト が可能になります。

Data Integration の オフライン同期 における embedding 機能は、2 つの設定モードをサポートしています。

ウィザードモードでの設定：ビジュアルインターフェイスを使用して、オフライン同期 を迅速に設定します。
スクリプトモードでの設定：このモードは詳細設定をサポートしており、カスタム要件を満たすためにさまざまな同期パイプラインを設定できます。

制限事項

この機能は、新バージョンのデータ開発が有効になっているワークスペースでのみ利用可能です。
この機能は、サーバーレスリソースグループのみをサポートしています。
この機能は現在、一部のオフライン同期チャネルでのみ利用可能です。

課金

Data Integration シナリオの標準料金に加えて、AI 支援 Data Integration タスクでは、大規模言語モデルの呼び出しにコストが発生します。

Alibaba Cloud DataWorks モデルサービスの課金については、「サーバーレスリソースグループの課金 — 大規模言語モデルサービス」をご参照ください。
Alibaba Cloud Model Studio の課金については、「モデル推論 (呼び出し) の課金」をご参照ください。
Alibaba Cloud PAI モデルマーケットプレイスの課金については、「Elastic Algorithm Service (EAS) の課金」をご参照ください。

前提条件

データ開発の新バージョンが有効化されたワークスペースを作成しました。
サーバーレスリソースグループを作成し、ワークスペースにバインド済みであること。
AI 支援処理に必要な大規模モデルサービスを設定済みであること。準備はモデルサービスプロバイダーによって異なります。
- Alibaba Cloud DataWorks モデルサービス：大規模モデルサービス管理でモデルをデプロイし、モデルサービスを開始済みであること。
- Alibaba Cloud Model Studio：Model Studio を有効化し、API キーを取得済みであること。
- Alibaba Cloud PAI モデルマーケットプレイス：PAI を有効化し、モデルサービストークンを取得済みであること。
オフライン同期タスクに必要なソースと宛先のデータソースを作成済みであること。

このチュートリアルでは、MaxCompute から Milvus へのデータ同期を例として使用します。したがって、まず MaxCompute データソースと Milvus データソースを作成する必要があります。

テストデータの準備

このチュートリアルでは、公開されているE コマース製品レビュー感情予測データセットのテーブルデータを使用します。ユーザーレビューをベクトル化し、類似性検索のために Milvus に同期します。

MaxCompute ソース：テストテーブルを作成し、テストデータを挿入します。

テストデータ

-- テストテーブルを作成します。
CREATE TABLE IF NOT EXISTS test_tb (
    sentence STRING,
    label STRING,
    dataset STRING
)
PARTITIONED BY (
    split STRING
)
LIFECYCLE 30;
-- テストデータを挿入します。
INSERT INTO test_tb PARTITION (split = 'dev')
SELECT * FROM VALUES
  ('Good for cleaning glass, but it is too small.', '1', 'jd'),
  ('The seller is so irresponsible. The quality of the clothes is terrible and not as shown in the picture.', '0', 'jd'),
  ('A great gift for international friends. Very nice!', '1', 'jd'),
  ('Very good. It will look beautiful once assembled.', '1', 'jd'),
  ('I returned the item to you, and you are trying to rip me off!', '0', 'jd'),
  ('Fast delivery. The book is genuine. JD.com is always my first choice for buying books!', '1', 'jd'),
  ('The taste is so good that I want to buy it again.', '1', 'jd'),
  ('The silicone smell is too strong, and it looks very different from the picture.', '0', 'jd'),
  ('So sad. I bought it for my Samsung N4, but it does not work. The customer service was unhelpful.', '0', 'jd'),
  ('The quality is good, the size is right, and it should be genuine! But I bought the black and gray one, and they sent me pure black. I am too lazy to exchange it, so I am leaving a bad review. Hope for improvement in the future!', '0', 'jd')
AS t (sentence, label, dataset);
-- データをクエリします。
SELECT * FROM test_tb WHERE split = 'dev';

Milvus 宛先：ベクトル化されたデータを受け取るための宛先コレクションを作成します。スキーマは以下の表で定義されています。

このコレクションでは自動 ID が有効になっています。

フィールド名	タイプ	説明
id	Int64	プライマリキー。自動インクリメント。
sentence	VarChar(32)	生テキストを格納します。
sentence_e	FloatVector(128)	類似性検索のためのベクトルフィールド。COSINE メジャーを使用します。

コードレス UI での設定

このチュートリアルでは、Data Integration のコードレス UI を使用してオフライン同期タスクを設定する方法を説明します。この例では、MaxCompute (ODPS) ソースからデータを読み取り、ベクトル埋め込みを生成し、データを Milvus 宛先に同期します。

ステップ 1：オフライン同期ノードの作成

DataWorks コンソールの [ワークスペース] ページに移動します。上部のナビゲーションバーで、目的のリージョンを選択します。目的のワークスペースを見つけ、[操作] 列で [ショートカット] > [Data Studio] を選択します。
プロジェクトディレクトリで、アイコンをクリックし、Create Node > Data Integration > Batch Synchronization を選択します。Source and Destination (このチュートリアルでは、ソースは MaxCompute、宛先は Milvus) を設定し、ノードの Name を入力して、OK をクリックします。

ステップ 2：オフライン同期タスクの設定

基本情報を設定します。
- Data Source：ソースと宛先のデータソースを選択します。
- Running Resources：オフライン同期タスクを実行するリソースグループを選択します。リソースグループは現在のワークスペースにバインドされ、データソースに接続されている必要があります。
利用可能なデータソースまたはリソースグループがない場合は、前提条件を完了していることを確認してください。

Data Source を設定します。

以下のパラメーターは、このチュートリアルで使用される MaxCompute データソースに適用されます。設定は他のデータソースによって異なるため、適宜設定を調整してください。

このチュートリアルでは、[テーブル] で test_tb を選択します。[フィルタリング方法] で [パーティションフィルター] を選択し、パーティション条件を split=dev に設定します。[パーティションが存在しない場合] で [無視してタスクを通常どおり実行] を選択します。

パラメーター	説明
Tunnel Resource Group	これはトンネルクォータのリソースグループです。デフォルト値は `Public Transmission Resource` で、これは MaxCompute の無料クォータです。MaxCompute のデータ転送リソースの選択に関する詳細については、「専用の Data Transmission Service リソースグループの購入と使用」をご参照ください。重要専用のトンネルクォータが有効期限切れや支払い遅延により利用できなくなった場合、実行中のタスクは自動的に `Public Transmission Resource` に切り替わります。
Tables	同期するソーステーブルを選択します。利用可能なソーステーブルがない場合は、テストデータの準備を完了していることを確認してください。
Filtering Method	Partition Filter と Data Filtering をサポートしています：パーティションテーブルの場合、パーティションを指定して同期するデータ範囲を選択できます。非パーティション化テーブルの場合、`WHERE` 句を使用して同期するデータをフィルタリングできます。

Data Preview をクリックして、設定が正しいかどうかを確認できます。

[データプレビュー] ダイアログボックスには、ソーステーブルのデータが表示されます。このチュートリアルでは、データには `sentence` (レビューテキスト)、`label` (感情ラベル)、`dataset` (データセットソース) の列が含まれます。

Data Processing を設定します。

データ処理を有効にします。Data Processing List で Add Node > [> Data Embedding] をクリックして [Data Embedding] 処理ノードを追加します。

データ埋め込みノードを設定します。[名前] に sentence2emb と入力します。[モデル名] で [text-embedding-v4] を選択します。ベクトル化するフィールドとして sentence を選択します。[ベクトル化出力フィールド] に sentence_e と入力します。[ベクトル次元] を 128 に設定し、[NULL を空文字列に変換] チェックボックスを選択します。以下の表に、主要なパラメーターを説明します。

説明

データ埋め込みノードのパフォーマンスは、設定されたモデルのパフォーマンスに依存します。Alibaba Cloud Model Studio が提供する QWen モデルには、秒間クエリ数 (QPS) の制限があります。Alibaba Cloud PAI モデルマーケットプレイスの場合、PAI-EAS 上にモデルをデプロイする必要があり、そのパフォーマンスはデプロイメントに使用されるリソース仕様に依存します。
特定のパラメーターセットに対して、埋め込みモデルは決定論的なベクトルを生成します。Data Integration は、同期中に LFU (Least Frequently Used) キャッシュを使用して、同一のソースデータに対する冗長なモデル呼び出しを回避します。これにより、処理パフォーマンスが向上し、埋め込みコストが削減されます。

パラメーター	説明
Model Provider	大規模モデルプロバイダー。以下のプロバイダーがサポートされています：Alibaba Cloud DataWorks モデルサービス、Alibaba Cloud Model Studio、Alibaba Cloud PAI モデルマーケットプレイス。
Model Name	埋め込みモデルの名前。ニーズに応じてモデルを選択します。
Model API Key	モデルへのアクセスに使用される API キー。モデルプロバイダーからキーを取得します。 Alibaba Cloud Model Studio：Model Studio API キーの取得。 Alibaba Cloud PAI モデルマーケットプレイス：デプロイされた EAS タスクに移動し、オンラインデバッグに入り、ヘッダーの `Authorization` パラメーターの値を取得します。この値を API キーとして使用します。
モデルエンドポイント	Model Providerとして[Alibaba Cloud PAI モデルマーケットプレイス]を選択した場合は、モデルエンドポイント (エンドポイント API アドレス) を設定する必要があります。
Batch Size	ベクトル化のバッチサイズ。これは、埋め込みモデルがバッチ処理をサポートしているかどうかによって異なります。バッチ処理により、埋め込みのパフォーマンスが向上し、コストが削減されます。デフォルト値は 10 です。
Select Fields to Vectorize	どの列をベクトル化するかを定義し、出力フィールドの名前を指定します。Data Integration は、単一のソースフィールドまたは複数の連結されたフィールドの組み合わせのベクトル化をサポートしています。
Vectorization Output Field	ベクトル化されたソースフィールドに定義されたベクトルフィールドの名前。
Vector Dimension	出力ベクトルの次元。埋め込みモデルはこの次元をサポートしている必要があります。デフォルト値は 1024 です。
Convert a NULL value to an empty string.	大規模モデルはベクトル化のために NULL 値を処理できません。このオプションを有効にすると、ソースデータ内の NULL が空文字列に変換され、エラーを防ぐことができます。このオプションはデフォルトで無効になっています。
Concatenate Field Names	ベクトル化のために、フィールド名とテキストを連結するかどうかを指定します。このオプションを有効にする場合は、 Field Name Delimiter も設定する必要があります。このオプションは、デフォルトでは無効です。
Skip Empty Fields	ベクトル化のために複数のフィールドを連結する際に、空のフィールドをスキップするかどうかを指定します。このオプションはデフォルトで有効になっています。

データ出力をプレビューします。

データ埋め込みノード設定エリアの右上隅にある Preview Data Output をクリックし、次に Preview をクリックして、ベクトル化された結果を表示し、設定を確認します。

オフライン同期エディターページの上部にある Dry Run をクリックして、ベクトル化された結果をプレビューすることもできます。

プレビュー結果では、新しい sentence_e 列に各文のベクトル埋め込み (浮動小数点数のシーケンス) が表示されます。

Destination を設定します。

以下のパラメーターは、このチュートリアルで使用される Milvus 宛先に適用されます。設定は他の宛先によって異なるため、適宜設定を調整してください。

このチュートリアルでは、[コレクション] を [Milvus_Collection] に設定し、[書き込みモード] で [insert] を選択し、[書き込みごとの行数] を [1024] に設定し、[コレクション作成モード] で [無視] を選択します。

パラメーター	説明
Collection	ベクトルデータを受け取るコレクション。
Partition Key	オプション。コレクションがパーティション化されている場合、受け取ったベクトルデータのパーティションを指定できます。
Write Mode	upsert：コレクションで自動 ID が無効になっている場合：プライマリキーに基づいてエンティティを更新します。コレクションで自動 ID が有効になっている場合：データを自動生成されたプライマリキーを持つ新しいエンティティとして挿入します。 insert：通常、自動 ID が有効になっているコレクションにデータを挿入するために使用され、Milvus が自動的にプライマリキーを生成します。自動 ID が有効になっていないコレクションで `insert` を使用すると、データが重複します。

Destination Field Mapping を設定します。

ソース、データ処理、送信先を設定すると、オフライン同期タスクによってフィールドマッピングが自動的に生成されます。送信先は非構造化スキーマであるため、タスクはデフォルトで、位置によってフィールドをマッピングします。 Edit を Source Field または Target Field の横でクリックして、マッピングの順序を調整したり、不要なフィールドを削除したりして、マッピングが正しいことを確認します。

たとえば、このチュートリアルでは、不要なフィールドを手動で削除します。調整後のマッピングは次のとおりです。

ソースフィールド sentence は宛先フィールド sentence (タイプ：VarChar、maxLength：32) にマッピングされ、ソースフィールド sentence_e は宛先フィールド sentence_e (タイプ：FloatVector、dimension：128) にマッピングされます。
Advanced Settings でその他のオプションを設定します。

ノード設定ページの右側にある Advanced Settings をクリックします。必要に応じて、タスクの同時実行数、同期レート、ダーティデータポリシーなどのパラメーターを設定できます。

ステップ 3：デバッグと実行

オフライン同期ノードエディターページの右側で、Run Configuration をクリックします。デバッグ実行のための Resource Group と Script Parameters を設定し、上部のツールバーで Running をクリックして同期パイプラインをテストします。
Milvus に移動し、宛先コレクションのデータが期待どおりであることを確認します。

ステップ 4：スケジューリングの設定と公開

オフライン同期タスクの右側にある Scheduling Settings をクリックします。スケジュール実行のためのスケジューリング設定パラメーターを設定し、上部のツールバーで Publish をクリックし、公開パネルの画面上のプロンプトに従って公開します。

スクリプトモードでの設定

このチュートリアルでは、Data Integration のスクリプトモードを使用してオフライン同期タスクを設定する方法を説明します。この例では、MaxCompute (ODPS) ソースからデータを読み取り、それをベクトル化して埋め込み、Milvus に同期します。

ステップ 1：オフライン同期ノードの作成

DataWorks コンソールの[ワークスペース] ページに移動します。上部ナビゲーションバーで、目的のリージョンを選択します。目的のワークスペースを見つけ、[操作] 列で[ショートカット] > [Data Studio] を選択します。
プロジェクトディレクトリで、アイコンをクリックし、Create Node > Data Integration > Batch Synchronization を選択します。Source and Destination (このチュートリアルでは、ソースは MaxCompute、宛先は Milvus) を設定し、ノードの Name を入力して、OK をクリックします。

同期スクリプトの設定

オフライン同期ノードのツールバーにあるアイコンをクリックして、スクリプトモードに切り替えます。

このタスクは、MaxCompute から Milvus にデータを同期します。

この例では、以下のスクリプトを使用します。スクリプトのフォーマットの詳細については、「付録 1：スクリプトモードのフォーマット仕様」をご参照ください。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "odps",
            "parameter": {
                "partition": [
                    "split=dev"
                ],
                "datasource": "MaxCompute_Source",
                "successOnNoPartition": true,
                "tunnelQuota": "default",
                "column": [
                    "sentence"
                ],
                "enableWhere": false,
                "table": "test_tb"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "category": "flatmap",
            "stepType": "embedding-transformer",
            "parameter": {
                "modelProvider": "bailian",
                "modelName": "text-embedding-v4",
                "embeddingColumns": {
                    "sourceColumnNames": [
                        "sentence"
                    ],
                    "embeddingColumnName": "sentence_e"
                },
                "apiKey": "sk-****",
                "dimension": 128,
                "nullAsEmptyString": true
            },
            "displayName": "sentence2emb",
            "description": ""
        },
        {
            "stepType": "milvus",
            "parameter": {
                "schemaCreateMode": "ignore",
                "enableDynamicSchema": true,
                "datasource": "Milvus_Source",
                "column": [
                    {
                        "name": "sentence",
                        "type": "VarChar",
                        "elementType": "None",
                        "maxLength": "32"
                    },
                    {
                        "name": "sentence_e",
                        "type": "FloatVector",
                        "dimension": "128",
                        "elementType": "None",
                        "maxLength": "65535"
                    }
                ],
                "writeMode": "insert",
                "collection": "Milvus_Collection",
                "batchSize": 1024,
                "columnMapping": [
                    {
                        "sourceColName": "sentence",
                        "dstColName": "sentence"
                    },
                    {
                        "sourceColName": "sentence_e",
                        "dstColName": "sentence_e"
                    }
                ]
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Reader と Writer のパラメーターの詳細については、「MaxCompute データソース」および「Milvus データソース」をご参照ください。
他のタイプのソースまたは宛先を使用する場合は、「サポートされているデータソースのリスト」をご参照ください。

以下の表は、データベクトル化スクリプトのパラメーターを説明しています。

パラメーター	説明	必須
modelProvider	大規模モデルプロバイダーを指定します。以下のプロバイダーがサポートされています。 dataworksModelService：DataWorks 大規模言語モデルサービスを通じてデプロイされたモデルサービス。 bailian：Alibaba Cloud Model Studio。QWen モデルをサポートしています。 paiModelGallery：Alibaba Cloud PAI モデルマーケットプレイス。BGE-M3 モデルをサポートしています。	はい
modelName	使用する埋め込みモデルの名前。 modelProvider が bailian の場合、`text-embedding-v4` または `text-embedding-v3` を選択できます。 modelProvider が paiModelGallery の場合、`bge-m3` を選択できます。	はい
apiKey	モデルへのアクセスに使用される API キー。モデルプロバイダーからキーを取得します。	はい
endpoint	モデルにアクセスするためのエンドポイント。これは、modelProvider が paiModelGallery に設定されている場合に必須です。	いいえ
batchSize	ベクトル化のバッチサイズ。これは、埋め込みモデルがバッチ処理をサポートしているかどうかによって異なります。バッチ処理により、埋め込みのパフォーマンスが向上し、コストが削減されます。デフォルト値は 10 です。	いいえ
embeddingColumns	ベクトル化する列と出力列の名前を定義します。Data Integration は、単一のソース列または複数の列の連結のベクトル化をサポートしています。例： `{ "embeddingColumns": { "sourceColumnNames": [ "col1", "col2" ], "embeddingColumnName": "my_vector" } }`	はい
appendDelimiter	複数の列の値を単一のテキスト文字列に結合してベクトル化するために使用される区切り文字。デフォルト：`\n`。	いいえ
skipEmptyValue	ベクトル化のために複数の列を連結する際に、空の列をスキップするかどうかを指定します。デフォルト：false。	いいえ
dimension	出力ベクトルの次元。埋め込みモデルはこの次元をサポートしている必要があります。デフォルト値は 1024 です。	いいえ
nullAsEmptyString	埋め込みモデルは NULL 入力を処理できません。ソースデータに NULL 値が含まれている場合、このパラメーターを `true` に設定して空文字列に変換し、ベクトル化エラーを防ぐことができます。デフォルト：`false`。	いいえ
appendFieldNameEnable	ベクトル化の前に列名をデータの先頭に追加するかどうかを指定します。`true` の場合、appendFieldNameDelimiter も設定する必要があります。デフォルト：`false`。	いいえ
appendFieldNameDelimiter	列名を連結するための区切り文字。このパラメーターは、appendFieldNameEnable が `true` に設定されている場合にのみ有効です。	いいえ

ノードのツールバーで Dry Run をクリックし、次に Start Collection と Preview をクリックして、ベクトル化された結果を確認し、設定を検証します。ダイアログボックスで、[サンプル数] を [3] に設定し、[サンプリングを開始] をクリックして入力データを取得します。入力データテーブルには `sentence`、`label`、`dataset` の列が含まれています。プレビュー結果には、新しい `sentence_e` 列が表示され、埋め込み変換後に各テキストエントリに対して生成された浮動小数点ベクトルが表示されます。入力データにダーティデータレコードは見つかりませんでした。結果が正しければ、[OK] をクリックします。
Advanced Settings でその他のオプションを設定します。

ノード設定ページの右側にある Advanced Settings をクリックします。必要に応じて、タスクの同時実行数、同期レート、ダーティデータポリシーなどのパラメーターを設定できます。

ステップ 3：デバッグと実行

オフライン同期ノードエディターページの右側で、Run Configuration をクリックします。デバッグ実行のための Resource Group と Script Parameters を設定し、上部のツールバーで Running をクリックして同期パイプラインをテストします。
Milvus に移動し、宛先コレクションのデータが期待どおりであることを確認します。

ステップ 4：スケジューリングの設定と公開

付録 1：スクリプトモードのフォーマット

スクリプトの基本構造は次のとおりです。

{
    "type": "job", 
    "version": "2.0", 
    "steps": [
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "transformer1",
            "category": "map/flatmap"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "transformer2",
            "category": "map/flatmap"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
    }
}

steps 配列は、各処理ステップを定義します。少なくとも 1 つの Reader と 1 つの Writer を含み、オプションで中間に複数の Transformer ステップを含めることができます。たとえば、同時実行数を 2 に設定すると、タスクには 2 つの並列データ処理ストリームができます。各 Reader、Transformer、Writer は、タスク設定の step と見なされます。

steps 配列は、各ステップのタイプとパラメーターを定義します。データ処理フローは、配列内の step 要素の順序に従います。

Data Integration がサポートするさまざまなデータソースの読み取り/書き込みコネクタの詳細なパラメーター設定については、「サポートされているデータソースと同期ソリューション」をご参照ください。

付録 2：OSS から Milvus への同期

この例では、OSS から JSON Lines (JSONL) データを同期し、それを解析し、指定されたフィールドをベクトル化し、出力を Milvus に書き込みます。以下の例は、完全な JSON 設定を示しています。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "oss",
            "parameter": {
                "datasource": "${YOUR_OSS_DATASOURCE_NAME}",
                "column": [
                    {
                        "name": "chunk_text",
                        "index": 0,
                        "type": "string"
                    }
                ],
                "fieldDelimiter": ",",
                "encoding": "UTF-8",
                "fileFormat": "jsonl",
                "object": [
                    "embedding/chunk1.jsonl"
                ]
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "json-extracting",
            "parameter": {
                "column": [
                    {
                        "name": "text",
                        "fromColumn": "chunk_text",
                        "jsonPath": "$.text",
                        "type": "STRING",
                        "nullOrInvalidDataAction": "DIRTY_DATA"
                    }
                ]
            },
            "name": "jsonextract",
            "category": "flatmap"
        },
        {
            "stepType": "embedding-transformer",
            "parameter": {
                "modelProvider": "bailian",
                "modelName": "text-embedding-v4",
                "apiKey": "${YOUR_API_KEY}",
                "embeddingColumns": {
                    "sourceColumnNames": [
                        "text"
                    ],
                    "embeddingColumnName": "my_vector"
                },
                "batchSize": 8,
                "dimension": 1024
            },
            "name": "embedding",
            "category": "flatmap"
        },
        {
            "stepType": "milvus",
            "parameter": {
                "schemaCreateMode": "ignore",
                "enableDynamicSchema": true,
                "datasource": "${YOUR_MILVUS_DATASOURCE_NAME}",
                "column": [
                    {
                        "name": "my_vector",
                        "type": "FloatVector",
                        "dimension": "1024",
                        "elementType": "None",
                        "maxLength": "65535"
                    },
                    {
                        "name": "text",
                        "type": "VarChar",
                        "elementType": "None",
                        "maxLength": "65535"
                    }
                ],
                "collection": "yunshi_vector_07171130",
                "writeMode": "insert",
                "batchSize": 1024,
                "columnMapping": [
                    {
                        "sourceColName": "my_vector",
                        "dstColName": "my_vector"
                    },
                    {
                        "sourceColName": "text",
                        "dstColName": "text"
                    }
                ]
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 1
        }
    }
}