メインコンテンツまでスキップ

Spark バッチジョブ

Spark バッチジョブを使用すると、Zilliz Cloud で管理される大規模データセットに対して分散オフライン処理を実行できます。組み込みジョブを使用して、ベクトルデータの重複排除、クラスター、検査を行えます。

Spark バッチジョブは、長時間実行されるデータ処理タスク向けに設計されています。低レイテンシーのオンラインリクエストやレコード単位の変換には適していません。

発生しうる課題​

ベクトルデータセットが大きくなるにつれて、単純な挿入や検索操作だけでは不十分になる場合があります。繰り返しの取り込みによる重複の発生、大規模な埋め込みコレクションの全体像の把握困難、前処理パイプラインの失敗による疑わしいレコードの残留などが生じることがあります。

重複するベクトル埋め込み​

再試行、繰り返しのインポート、データソースの重複、同じテキストや画像の微修正版などが原因で、エンティティ間で重複するベクトル埋め込みが生成されることがあります。同じプライマリキーを持つエンティティだけでなく、プライマリキーは異なるものの内容がほぼ同一のエンティティも存在し、ストレージ、インデックス、後続処理のコストを増加させます。

Spark バッチジョブは、大規模データセット全体から重複を特定して削除できます。同じ ID を持つレコードを整理するには プライマリキー重複排除ジョブ を、ID は異なるが内容がほぼ同一のレコードを検出するには ベクトル類似度重複排除ジョブ を使用します。

不明瞭な埋め込み分布​

コレクションが大きくなるにつれて、埋め込み分布の全体像を把握するのが難しくなります。データセット内でどのパターンが支配的か、ロングテールデータがどこに分布しているか、新規インポートデータが既存レコードと異なっているかなどが不明確になる場合があります。

K-Means クラスタリングジョブ は、類似する埋め込みを大まかなクラスターにグループ化し、各レコードにクラスター ID を割り当てます。この結果を用いて、データ分布の分析、データソースの比較、代表的なサンプルの作成、類似度ベースの処理の細分化などを行えます。

埋め込みデータに潜む異常​

埋め込みパイプラインは、一見正常でもデータセットの他の部分と一致しないレコードを生成することがあります。前処理の失敗、不適切なモデル、解析ノイズ、破損したソースコンテンツ、想定外のデータバッチなどが原因で、手動での確認では発見しにくい異常な埋め込みが生じる可能性があります。

異常検出ジョブ は、埋め込み分布をスキャンし、一般的なパターンから大きく外れたレコードを特定します。この結果を活用して、再埋め込み、クリーンアップ、詳細なレビューが必要なデータを特定できます。異常が直ちに無効なデータを意味するわけではないため、フラグ付きレコードは自動削除せず、必ずレビューを行ってください。

ジョブタイプの選択​

次の表に、目的に応じた推奨ジョブタイプを示します。

目的推奨ジョブ
同じプライマリキーを持つレコードの削除プライマリキー重複排除
ベクトル表現が非常に類似しているレコードの検出ベクトル類似度重複排除
ベクトルデータを指定した数のグループに分割K-Means クラスタリング
主要なデータ分布から大きく外れたレコードの検出異常検出

Spark バッチジョブの仕組み​

Spark バッチジョブは、長時間実行される分散オフライン処理ジョブであり、ジョブ作成リクエストを受け取るとすぐにジョブ ID を返します。このジョブ ID を使用して、進行状況の監視やライフサイクルの管理を行えます。

事前準備​

Spark バッチジョブを送信する前に、以下の条件を満たしていることを確認してください。

  • 十分な権限を持つ有効な Zilliz Cloud API キーを保有していること。

  • 入力データが、サポートされている形式で Zilliz Cloud Volume に配置されていること。

    • サポートされているデータファイル形式は、parquet、lance、json、csv です。

  • 各 External Volume に関連付けられたストレージロールに、ジョブの実行に必要な権限が付与されていること。

    • 入力 External Volume には、入力先への読み取りアクセスが必要です。

    • 出力 External Volume には、出力先への読み書きアクセスに加え、ジョブ実行中に作成された一時オブジェクトや不完全なオブジェクトを削除する権限が必要です。

    • 入力用と出力用の External Volume で、異なるストレージロールを使用することも可能です。

External Volume の権限設定​

Spark バッチジョブは、External Volume から入力データを読み取り、結果を External Volume に書き込みます。ジョブの送信前に、各 Volume に関連付けられたオブジェクトストレージロールに、用途に応じた適切な権限が付与されていることを確認してください。

入力用と出力用の Volume で、異なるストレージロールを使用できます。各ロールには、対応するバケットおよびプレフィックスに対して必要な権限のみを付与してください。

入力 Volume​

入力 Volume には、入力データへの読み取りアクセスが必要です。Volume が使用する Integration に必要な読み取り権限が既に含まれている場合は、追加の設定は不要です。

Amazon S3 の場合、ロールには以下の権限が必要です。

  • 入力プレフィックス配下のオブジェクトに対する s3:GetObject

  • 入力プレフィックス配下のオブジェクトを一覧表示するための s3:ListBucket

  • バケットに対する s3:GetBucketLocation

例を表示するにはここをクリックしてください。
json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListInput",
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": "arn:aws:s3:::<bucket>",
      "Condition": {
        "StringLike": {
          "s3:prefix": [
            "<input-prefix>",
            "<input-prefix>/*"
          ]
        }
      }
    },
    {
      "Sid": "ReadInput",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::<bucket>/<input-prefix>/*"
    }
  ]
}

出力 Volume​

出力 Volume には、ジョブ結果を書き込むための権限が必要です。また、Spark が書き込みの失敗やキャンセルによって残された一時オブジェクトなどを削除する場合もあります。

Amazon S3 の場合、ロールには以下の権限が必要です。

  • 出力先にアクセスするための s3:GetObject、s3:ListBucket、s3:GetBucketLocation

  • 結果を書き込むための s3:PutObject

  • 一時オブジェクトや不完全な出力オブジェクトをクリーンアップするための s3:DeleteObject

s3:PutObject および s3:DeleteObject のスコープは、バケット全体ではなく出力プレフィックスに限定してください。

例を表示するにはここをクリックしてください。
json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListOutput",
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": "arn:aws:s3:::<bucket>",
      "Condition": {
        "StringLike": {
          "s3:prefix": [
            "<output-prefix>",
            "<output-prefix>/*"
          ]
        }
      }
    },
    {
      "Sid": "ReadWriteOutput",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject"
      ],
      "Resource": "arn:aws:s3:::<bucket>/<output-prefix>/*"
    }
  ]
}

Spark バッチジョブの送信と実行​

API キーを取得し、必要なファイルを Zilliz Cloud ボリュームにアップロードしたら、Spark バッチジョブを送信・実行する準備は完了です。

1

べき等性キーを生成します。

べき等性キーとは、同じジョブリクエストを再試行する際に変わらない一意の文字列です。詳細については、べき等性のある送信 を参照してください。

2

リクエストヘッダーを準備します。

Spark バッチジョブを作成する際は、前の手順で生成したべき等性キーをリクエストヘッダーに含めます。

http
Authorization: Bearer <api-key>
Idempotency-Key: spark-job-20260730-001
Content-Type: application/json
3

リクエストペイロードを準備します。

すべての Spark バッチジョブは共通のペイロード構造を持ちますが、ジョブ固有のパラメーターは目的に応じて異なります。

共通のペイロード構造については リクエストペイロード を参照してください。ジョブ固有のパラメーターについては、以下のページを参照してください。

べき等性のある送信​

べき等性キーを使用することで、ジョブ送信を安全に再試行できます。Zilliz Cloud はキーとリクエストボディの両方を確認し、既存のジョブを返すか、競合としてリクエストを拒否するかを判断します。照合動作の概要を次の表に示します。

ケース動作
同じべき等性キーかつ同じリクエストボディ重複を作成せず、以前に作成された Spark バッチジョブを返します。
同じべき等性キーだが異なるリクエストボディ競合エラーを返します。

べき等性キーのスコープは、特定の組織、ユーザー、リージョン、プロジェクトに限定されます。Zilliz Cloud は、ジョブの最大タイムアウト時間に基づき、キーを約 25 時間 保持します。保持期間が過ぎた後は、同じキーを新しい送信に再利用できます。

リクエストペイロード​

すべての Spark バッチジョブは、以下のような共通のペイロード構造を持ちます。

json
{
  "description": "optional description",
  "regionId": "aws-us-west-2",
  "input": {...},
  "output": {...},
  "resourceSize": "SMALL",
  "timeoutSeconds": 3600
}

各パラメーターの説明を次の表に示します。

パラメーター必須説明
description任意ジョブに関する任意の説明です。
値は 1,024 文字以内の文字列です。
regionId必須ジョブを実行する Zilliz Cloud リージョンの ID です。サポートされているリージョンについては、Cloud Providers & Regions を参照してください。
input任意ジョブの入力です。組み込みジョブの場合のみ必須となります。詳細については、下の表を参照してください。
output任意ジョブの出力です。組み込みジョブの場合は必須です。詳細については、下の表を参照してください。
resourceSize任意ジョブに必要な Spark クラスターのサイズです。指定可能な値は SMALL、MEDIUM、LARGE、XLARGE、2XLARGE、3XLARGE です。
timeoutSeconds任意ジョブのタイムアウト時間（秒）です。値は 300 から 86400 の範囲の正の整数です。

上記の表の input および output パラメーターは、以下のような同様の構造を持ちます。

パラメーター

必須

説明

type

はい

Spark バッチジョブのタイプです。input と output の両方に適用されます。指定可能な値は次のとおりです。

  • volume

volumeName

いいえ

Zilliz Cloud ボリュームの名前です。type を volume に設定する場合に必須となります。input と output の両方に適用されます。

path

いいえ

指定した Zilliz Cloud ボリュームのルートからの相対パスで指定する入力/outputファイルパスです。type を volume に設定する場合に必須となります。input と output の両方に適用されます。

例えば、ファイルが volume://path/to/data.parquet にある場合、path を path/to/data.parquet に設定します。

format

いいえ

入力または出力ファイルの形式です。input と output の両方に適用されます。デフォルト値は parquet です。指定可能な値は parquet、lance、json、csv です。

writeMode

いいえ

出力ファイルの書き込みモードです。output にのみ適用されます。指定可能な値は次のとおりです。

  • ERROR_IF_EXIST

    指定した出力ファイルが既に存在する場合にエラーを返します。これがデフォルトのオプションです。

  • OVERWRITE

    指定したファイルを上書きします。

📘Notes

input では、format によって使用される Spark データソースリーダーが決まります。このパラメーターを省略した場合、ジョブはデフォルトで Parquet リーダーを使用し、指定したパス配下の Parquet ファイルのみを処理します。JSON や CSV など他の形式のファイルは無視されます。

処理対象のファイルが Parquet 以外の場合は、input.format に対応する形式を明示的に設定してください。ジョブは異なる形式のファイルを自動的に検出・結合しません。

送信レスポンス​

ジョブリクエストが成功した場合でも、返される HTTP コードは異なることがあります。レスポンスに含まれる HTTP コードを次の表に示します。

ケースHTTP コード説明
ジョブ作成時201 CREATEDジョブが作成中であることを示します。
ジョブ作成は非同期で行われるため、レスポンスに含まれるジョブ ID を使用して進行状況の確認やライフサイクル管理を行えます。
ジョブキャンセル時202 ACCEPTEDキャンセル要求が受け付けられ、処理中であることを示します。
ジョブのキャンセルは非同期で行われるため、レスポンスに含まれるジョブ ID を使用して進行状況を確認できます。
ジョブ情報取得時200 OKリクエストに対するレスポンスが正常に返されたことを示します。
これは同期処理であり、レスポンスにはリクエスト処理時点のジョブステータスが常に含まれます。

HTTP コードは異なりますが、ペイロード構造は以下のように共通です。

json
{
  "code": 0,
  "data": {
    "jobId": "job-xxxxxxxx",
    "projectId": "proj-xxxxxxxx",
    "type": "SPARK",
    "description": "backfill product attributes",
    "status": "PENDING",
    "regionId": "aws-us-west-2",
    "clusterId": "in-xxxxxxxx",
    "createdAt": null,
    "startedAt": null,
    "finishedAt": null,
    "durationSeconds": null
  }
}

レスポンスにはジョブの監視に使用できるジョブ ID が含まれます。ジョブの状態確認、詳細情報の取得、実行中のジョブのキャンセルについては、ジョブステータスの理解 を参照してください。

エラーが発生した場合、レスポンスは以下のようになります。

json
{
  "code": 10001,
  "message": "projectId is required",
  "details": {
    "errorCode": "INVALID_PARAMETER"
  }
}

エラーの内容は、details.errorCode とエラーレスポンスに含まれる HTTP コードから特定できます。該当する HTTP コードとその意味を次の表に示します。

HTTP コード説明
400 BAD REQUESTリクエストに含まれるパラメーターが不正であることを示します。
403 FORBIDDENAPI キーに十分な権限がないか、指定したプロジェクトでリソースを利用できないことを示します。
404 NOT FOUNDジョブ ID や Zilliz Cloud ボリュームなど、指定されたリソースが存在しないことを示します。
409 CONFLICTべき等性キーがリクエストペイロードと一致しないことを示します。
500 INTERNAL SERVER ERRORサーバー側でリクエストの処理に失敗したことを示します。

次のステップ​

以下のガイドを参照して、目的に合った Spark バッチジョブの作成や、既存ジョブの監視・管理を行ってください。

Primary-Key Deduplication

Primary-key deduplication identifies records that share the same primary key and removes redundant copies from a large dataset. Use this job to clean up duplicates introduced by repeated imports, pipeline retries, migrations, or overlapping data sources. | Cloud

Vector Similarity Deduplication

Vector similarity deduplication identifies records with highly similar embeddings and groups them as semantic duplicates. Use this job to reduce semantic redundancy, such as paraphrased text, slightly modified images, or multiple versions of similar content. | Cloud

K-Means クラスタリング

K-Means クラスタリングは、類似した埋め込みを持つレコードを指定された数のクラスターにグループ化します。このジョブを使用すると、ベクトルデータの分布を把握したり、レコードを大まかな意味グループに整理したり、サンプリングや分析などの後続ワークフロー向けにデータセットを準備できます。 | Cloud

異常検知

異常検知は、ベクトル埋め込みが全体的なデータ分布から大きく逸脱しているレコードを特定します。このジョブを使用すると、データ品質の問題、まれなケース、処理エラー、または追加の確認が必要なサンプルを示唆する異常なレコードを検出できます。 | Cloud

データバックフィル

データバックフィルを使用すると、Zilliz Cloud コレクション内の既存エンティティの特定のフィールドを、Zilliz Cloud ボリューム上の Parquet ファイルに保存されたデータを使って更新できます。バックフィルジョブは、主キーに基づいて入力レコードと既存エンティティを照合し、指定されたフィールド値をコレクションに書き戻します。新規追加フィールドへの値の設定、欠損値の補完、既存フィールド値の一括置換などに利用できます。 | Cloud

Spark バッチジョブの管理

Spark バッチジョブは非同期で実行され、送信から完了まで複数の状態を遷移します。このページでは、ジョブのライフサイクルについて説明し、ジョブの一覧表示、詳細の取得、およびキャンセル可能な状態にあるジョブのキャンセル方法をご紹介します。 | Cloud

Ctrl I