DataWorks での OpenSearch へのオフラインデータ書き込み - DataWorks

DataWorks Data Integration の OpenSearch Writer プラグインを使用して、OpenSearch にデータを書き込むことができます。このトピックでは、オフラインモードで OpenSearch にデータを書き込む方法について説明します。

サポートされているバージョン

バージョン 3 はセカンドパーティパッケージを使用します。pom 依存関係は com.aliyun.opensearch:aliyun-sdk-opensearch:2.1.3 です。
OpenSearch Writer プラグインを使用するには、JDK 1.6-32 以降のバージョンが必要です。java -version コマンドを実行して、Java のバージョン番号を確認できます。
Alibaba Cloud OpenSearch の次の商用エディションがサポートされています: Industry Algorithm Edition、LLM ベースの AI チャットエディション、高性能検索エディション、Vector Search Edition、検索エンジンエディション。

制限事項

OpenSearch Writer は、サーバーレスリソースグループ (推奨) および Data Integration 専用リソースグループをサポートしますが、カスタムリソースグループはサポートしません。
OpenSearch の列は順序付けられていません。したがって、OpenSearch Writer では、データを書き込む列の順序を指定する必要があります。宛先の OpenSearch テーブルの列数よりも少ない列を指定した場合、指定されていない列はデフォルト値または null に設定されます。
たとえば、OpenSearch テーブルに列 a、b、c が含まれており、列 b と c にデータをインポートする場合、"column":["c","b"] パラメーターを設定できます。この構成では、リーダーの 1 番目と 2 番目の列が、それぞれ OpenSearch テーブルの列 c と b にインポートされます。列 a はデフォルト値または null に設定されます。
オフラインデータを OpenSearch に書き込むことができるのは、[コードエディタ] モードのみです。

サポートされているフィールドタイプ

OpenSearch Writer は、ほとんどの OpenSearch データ型をサポートしています。次の表に、サポートされているデータ型のマッピングを示します。

カテゴリ	OpenSearch データ型
整数	INT
浮動小数点	DOUBLE および FLOAT
文字列	TEXT、LITERAL、および SHORT_TEXT
日付と時刻	INT
ブール値	LITERAL

データ同期タスクの開発

同期タスクを設定するためのエントリポイントと手順については、次の設定ガイドをご参照ください。

手順については、「コードエディタモードでの設定」をご参照ください。
コードエディタモードのすべてのパラメーターとコードサンプルについては、「付録: コードサンプルとパラメーター」をご参照ください。

よくある質問

列設定エラーの処理

データの信頼性を確保するため、OpenSearch Writer は列数を検証します。宛先テーブルに存在する列よりも多くの列を書き込もうとすると、ライターはエラーを報告します。たとえば、OpenSearch テーブルに列 a、b、c がある場合、3 つを超える列を書き込もうとすると、OpenSearch Writer はエラーを報告します。

テーブル設定に関する注意事項

OpenSearch Writer は、一度に 1 つのテーブルにのみデータを書き込むことができます。

タスクの再実行とフェールオーバー

タスクが再実行されると、既存のデータはドキュメント ID に基づいて上書きされます。したがって、OpenSearch に書き込む列には、行の一意の識別子として機能する ID 列を含める必要があります。ID が一致するデータは上書きされます。

付録: コードサンプルとパラメーター

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合は、統一されたスクリプトフォーマットの要件に基づいて、スクリプト内の関連パラメーターを設定する必要があります。詳細については、「コードエディタでタスクを設定する」をご参照ください。次の情報は、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明しています。

Writer のコードサンプル (Industry Algorithm Edition、LLM ベースの AI チャットエディション、高性能検索エディション)

{
    "type": "job",
    "version": "1.0",
    "configuration": {
        "reader": {},
        "writer": {
            "plugin": "opensearch",
            "parameter": {
                "accessId": "*********",
                "accessKey": "********",
                "host": "http://yyyy.aliyuncs.com",
                "endpoint":"http://yyyy.aliyuncs.com",
                "indexName": "datax_xxx",
                "table": "datax_yyy",
                "column": [
                "appkey",
                "id",
                "title",
                "gmt_create",
                "pic_default"
                ],
                "batchSize": 500,
                "writeMode": add,
                "version":"v2",
                "ignoreWriteError": false
            }
        }
    }
}

Writer のパラメーター (Industry Algorithm Edition、LLM ベースの AI チャットエディション、高性能検索エディション)

パラメーター	説明	必須	デフォルト値
accessId	AccessKey ペアの AccessKey ID です。	はい	N/A
accessKey	AccessKey ペアの AccessKey Secret です。ログインパスワードとして使用されます。	はい	N/A
host	OpenSearch のトラフィックドメイン名です。OpenSearch コンソールにログインし、インスタンス詳細ページに移動してドメイン名を取得できます。	はい	N/A
endpoint	OpenSearch の管理エンドポイントです。対応する OpenSearch エディションの公式サイトからエンドポイントを取得できます。たとえば、Industry Algorithm Edition の場合は、「サービスエンドポイント」をご参照ください。	はい	N/A
indexName	OpenSearch プロジェクトの名前です。	はい	N/A
table	宛先テーブルの名前です。DataX は複数のテーブルへの同時書き込みをサポートしていないため、テーブルは 1 つしか指定できません。	はい	N/A
column	データを書き込む列です。すべての列にデータを書き込むには、このパラメーターを `"column":["*"]` に設定します。特定の列にデータを書き込むには、`"column":["id","name"]` のように列名を指定します。 OpenSearch は、列のフィルタリングと並べ替えをサポートしています。たとえば、テーブルに列 a、b、c があるとします。列 c と b にのみデータを同期する場合は、このパラメーターを `["c","b"]` に設定できます。インポート中、列 a は自動的に null に設定されます。	はい	N/A
batchSize	各バッチで書き込むデータレコードの数です。OpenSearch はバッチ書き込みを実行します。OpenSearch の主な強みはクエリにあり、書き込みの秒間トランザクション数 (TPS) は高くありません。アカウントに割り当てられたリソースに基づいて、このパラメーターを設定してください。通常、単一のデータレコードは 1 MB 未満で、単一のバッチ書き込みは 2 MB 未満です。	このパラメーターはパーティションテーブルに必須です。非パーティションテーブルにはこのパラメーターを指定しないでください。	300
writeMode	書き込みモードです。書き込みのべき等性を確保するために "writeMode":"add/update" を設定します: "add": 書き込みに失敗してタスクが再実行されると、OpenSearch Writer はデータをクリアして新しいデータをインポートします。これはアトミック操作です。 "update": データは更新として挿入されます。これはアトミック操作です。説明 OpenSearch でのバッチ挿入はアトミック操作ではありません。一部のレコードは正常に挿入されても、他のレコードは失敗する可能性があります。writeMode パラメーターの選択は重要です。更新操作はバージョン 3 ではサポートされていません。	はい	N/A
ignoreWriteError	書き込みエラーを無視できます。例: `"ignoreWriteError":true`。OpenSearch はバッチ書き込みを実行します。このパラメーターは、現在のバッチの書き込み失敗を無視するかどうかを指定します。失敗を無視すると、他の書き込み操作は続行されます。失敗を無視しない場合、現在のタスクは停止し、エラーを返します。デフォルト値を使用することをお勧めします。	いいえ	false
version	OpenSearch のバージョンです (例: `"version":"v3"`)。プッシュ操作にはバージョン 2 では多くの制限があるため、バージョン 3 を使用することをお勧めします。	いいえ	v2

Writer のコードサンプル (Vector Search Edition および検索エンジンエディション)

{
  "stepType": "opensearch",
  "parameter": {
    "indexName": "",
    "column": [
      {
        "name": "col3double",
        "type": "DOUBLE"
      },
      {
        "name": "col2vector",
        "type": "MULTI_FLOAT"
      }
    ],
    "datasource": "zm_test_vector_01",
    "batchSize": "500",
    "table": "demotable"
  },
  "name": "Writer",
  "category": "writer"
}

Writer のパラメーター (Vector Search Edition および検索エンジンエディション)

パラメーター	説明	必須	デフォルト値
table	宛先テーブルの名前です。DataX は複数のテーブルへの同時書き込みをサポートしていないため、テーブルは 1 つしか指定できません。	はい	N/A
column	データを書き込む列です。すべての列にデータを書き込むには、このパラメーターを `"column":["*"]` に設定します。特定の列にデータを書き込むには、`"column":["id","name"]` のように列名を指定します。 OpenSearch は、列のフィルタリングと並べ替えをサポートしています。たとえば、テーブルに列 a、b、c があるとします。列 c と b にのみデータを同期する場合は、このパラメーターを `["c","b"]` に設定できます。インポート中、列 a は自動的に null に設定されます。	はい	N/A
batchSize	各バッチで書き込むデータレコードの数です。OpenSearch はバッチ書き込みを実行します。OpenSearch の主な強みはクエリにあり、書き込みの秒間トランザクション数 (TPS) は高くありません。アカウントに割り当てられたリソースに基づいて、このパラメーターを設定してください。通常、単一のデータレコードは 1 MB 未満で、単一のバッチ書き込みは 2 MB 未満です。	このパラメーターはパーティションテーブルに必須です。非パーティションテーブルにはこのパラメーターを指定しないでください。	300