Azure AI Search プッシュ API を使用してデータのインデックスを作成する

REST API は、Azure AI Search インデックスにデータをプッシュする最も柔軟な方法です。 任意のプログラミング言語を使用することも、エンドポイントに JSON 要求を投稿できる任意のアプリで対話形式で使用することもできます。

ここでは、REST API を効果的に使用し、使用可能な操作を調べる方法について説明します。 次に、.NET Core コードを見て、API を使用して大量のデータの追加を最適化する方法を確認します。

サポートされている REST API 操作

AI Search でサポートされている REST API は 2 つあります。 検索 API と管理 API。 このモジュールでは、検索の 5 つの機能に対する操作を提供する検索 REST API に重点を置いています。

検索 REST API を呼び出す方法

いずれかの検索 API を呼び出す場合は、次の手順を実行する必要があります。

• 検索サービスによって提供される HTTPS エンドポイント (既定のポート 443 経由) を使用します。URI に api バージョンの を含める必要があります。
• 要求ヘッダーには、api-key 属性を含める必要があります。

エンドポイント、api-version、api-key を検索するには、Azure portal に移動します。

ポータルで検索サービスに移動し、検索エクスプローラー 選択します。 REST API エンドポイントは、要求 URL フィールドにあります。 URL の最初の部分はエンドポイント (https://azsearchtest.search.windows.netなど) であり、クエリ文字列には api-version が表示されます (例: api-version=2023-07-01-Preview)。

左側の api-key を見つけるには、[キー 選択します。 REST API を使用してインデックスのクエリを実行する以外の操作を行う場合は、プライマリまたはセカンダリ管理者キーを使用できます。 インデックスを検索するだけで、クエリ キーを作成して使用できます。

インデックス内のデータを追加、更新、または削除するには、管理者キーを使用する必要があります。

インデックスにデータを追加する

次の形式のインデックス機能を使用して HTTP POST 要求を使用します。

POST https://[service name].search.windows.net/indexes/[index name]/docs/index?api-version=[api-version] 

要求の本文では、ドキュメントに対して実行するアクション、アクションを適用するドキュメント、および使用するデータを REST エンドポイントに通知する必要があります。

JSON は次の形式である必要があります。

{  
  "value": [  
    {  
      "@search.action": "upload (default) | merge | mergeOrUpload | delete",  
      "key_field_name": "unique_key_of_document", (key/value pair for key field from index schema)  
      "field_name": field_value (key/value pairs matching index schema)  
        ...  
    },  
    ...  
  ]  
}

要求が成功すると、API は 200 状態コードを返します。

次の例の JSON は、前のユニットの顧客レコードをアップロードします。

{
  "value": [
    {
      "@search.action": "upload",
      "id": "5fed1b38309495de1bc4f653",
      "firstName": "Sims",
      "lastName": "Arnold",
      "isAlive": false,
      "age": 35,
      "address": {
        "streetAddress": "Sumner Place",
        "city": "Canoochee",
        "state": "Palau",
        "postalCode": "1558"
      },
      "phoneNumbers": [
        {
          "phoneNumber":  {
            "type": "home",
            "number": "+1 (830) 465-2965"
          }
        },
        {
          "phoneNumber":  {
            "type": "home",
            "number": "+1 (889) 439-3632"
          }
        }
      ]
    }
  ]
}

値配列には、必要な数のドキュメントを追加できます。 ただし、最適なパフォーマンスを得るために、最大 1,000 個のドキュメントまたは合計サイズで 16 MB までのドキュメントをバッチ処理することを検討してください。

.NET Core を使用してデータのインデックスを作成する

最適なパフォーマンスを得るための最新の Azure.Search.Document クライアント ライブラリ (現在はバージョン 11) を使用します。 NuGet を使用してクライアント ライブラリをインストールできます。

dotnet add package Azure.Search.Documents --version 11.4.0

インデックスの実行方法は、次の 6 つの重要な要因に基づいています。

• 検索サービス レベルと、有効にしたレプリカとパーティションの数。
• インデックス スキーマの複雑さ。 各フィールドに含まれるプロパティ (検索可能、ファセット可能、並べ替え可能) の数を減らします。
• 各バッチ内のドキュメントの数、最適なサイズは、インデックス スキーマとドキュメントのサイズによって異なります。
• アプローチをマルチスレッド化する方法。
• エラーと調整の処理。 指数バックオフ再試行戦略を使用します。
• データが存在する場合は、検索インデックスの近くにデータのインデックスを作成してみてください。 たとえば、Azure 環境内からアップロードを実行します。

最適なバッチ サイズを確認する

最適なバッチ サイズを解決することは、パフォーマンスを向上させるための重要な要素であるため、コード内のアプローチを見てみましょう。

public static async Task TestBatchSizesAsync(SearchClient searchClient, int min = 100, int max = 1000, int step = 100, int numTries = 3)
{
    DataGenerator dg = new DataGenerator();

    Console.WriteLine("Batch Size \t Size in MB \t MB / Doc \t Time (ms) \t MB / Second");
    for (int numDocs = min; numDocs <= max; numDocs += step)
    {
        List<TimeSpan> durations = new List<TimeSpan>();
        double sizeInMb = 0.0;
        for (int x = 0; x < numTries; x++)
        {
            List<Hotel> hotels = dg.GetHotels(numDocs, "large");

            DateTime startTime = DateTime.Now;
            await UploadDocumentsAsync(searchClient, hotels).ConfigureAwait(false);
            DateTime endTime = DateTime.Now;
            durations.Add(endTime - startTime);

            sizeInMb = EstimateObjectSize(hotels);
        }

        var avgDuration = durations.Average(timeSpan => timeSpan.TotalMilliseconds);
        var avgDurationInSeconds = avgDuration / 1000;
        var mbPerSecond = sizeInMb / avgDurationInSeconds;

        Console.WriteLine("{0} \t\t {1} \t\t {2} \t\t {3} \t {4}", numDocs, Math.Round(sizeInMb, 3), Math.Round(sizeInMb / numDocs, 3), Math.Round(avgDuration, 3), Math.Round(mbPerSecond, 3));

        // Pausing 2 seconds to let the search service catch its breath
        Thread.Sleep(2000);
    }

    Console.WriteLine();
}

この方法では、バッチ サイズを大きくし、有効な応答の受信にかかる時間を監視します。 コードは 100 から 1000 まで、100 のドキュメント ステップでループします。 バッチ サイズごとに、ドキュメント サイズ、応答を取得する時間、および MB あたりの平均時間が出力されます。 このコードを実行すると、次のような結果が得られます。

上記の例では、スループットに最適なバッチ サイズは 1 秒あたり 2.499 MB、バッチあたり 800 ドキュメント。

エクスポネンシャル バックオフの再試行戦略を実装する

オーバーロードが原因でインデックスが要求のスロットルを開始すると、503 (負荷が高いため要求が拒否されました) または 207 (一部のドキュメントがバッチで失敗しました) 状態で応答します。 これらの応答を処理する必要があります。適切な戦略はバックオフです。 バックオフとは、要求を再試行する前にしばらく一時停止することを意味します。 エラーごとにこの時間を増やすと、指数関数的にバックオフされます。

次のコードを確認します。

// Implement exponential backoff
do
{
    try
    {
        attempts++;
        result = await searchClient.IndexDocumentsAsync(batch).ConfigureAwait(false);

        var failedDocuments = result.Results.Where(r => r.Succeeded != true).ToList();

        // handle partial failure
        if (failedDocuments.Count > 0)
        {
            if (attempts == maxRetryAttempts)
            {
                Console.WriteLine("[MAX RETRIES HIT] - Giving up on the batch starting at {0}", id);
                break;
            }
            else
            {
                Console.WriteLine("[Batch starting at doc {0} had partial failure]", id);
                Console.WriteLine("[Retrying {0} failed documents] \n", failedDocuments.Count);

                // creating a batch of failed documents to retry
                var failedDocumentKeys = failedDocuments.Select(doc => doc.Key).ToList();
                hotels = hotels.Where(h => failedDocumentKeys.Contains(h.HotelId)).ToList();
                batch = IndexDocumentsBatch.Upload(hotels);

                Task.Delay(delay).Wait();
                delay = delay * 2;
                continue;
            }
        }

        return result;
    }
    catch (RequestFailedException ex)
    {
        Console.WriteLine("[Batch starting at doc {0} failed]", id);
        Console.WriteLine("[Retrying entire batch] \n");

        if (attempts == maxRetryAttempts)
        {
            Console.WriteLine("[MAX RETRIES HIT] - Giving up on the batch starting at {0}", id);
            break;
        }

        Task.Delay(delay).Wait();
        delay = delay * 2;
    }
} while (true);

このコードは、失敗したドキュメントをバッチで追跡します。 エラーが発生した場合は、遅延を待機し、次のエラーの遅延を 2 倍にします。

最後に、再試行の最大数があり、この最大数に達した場合はプログラムが存在します。

スレッド処理を使用してパフォーマンスを向上させる

上記のバックオフ戦略とスレッド化アプローチを組み合わせて、ドキュメントアップロードアプリを完成させることができます。 次にコード例をいくつか示します。

        public static async Task IndexDataAsync(SearchClient searchClient, List<Hotel> hotels, int batchSize, int numThreads)
        {
            int numDocs = hotels.Count;
            Console.WriteLine("Uploading {0} documents...\n", numDocs.ToString());

            DateTime startTime = DateTime.Now;
            Console.WriteLine("Started at: {0} \n", startTime);
            Console.WriteLine("Creating {0} threads...\n", numThreads);

            // Creating a list to hold active tasks
            List<Task<IndexDocumentsResult>> uploadTasks = new List<Task<IndexDocumentsResult>>();

            for (int i = 0; i < numDocs; i += batchSize)
            {
                List<Hotel> hotelBatch = hotels.GetRange(i, batchSize);
                var task = ExponentialBackoffAsync(searchClient, hotelBatch, i);
                uploadTasks.Add(task);
                Console.WriteLine("Sending a batch of {0} docs starting with doc {1}...\n", batchSize, i);

                // Checking if we've hit the specified number of threads
                if (uploadTasks.Count >= numThreads)
                {
                    Task<IndexDocumentsResult> firstTaskFinished = await Task.WhenAny(uploadTasks);
                    Console.WriteLine("Finished a thread, kicking off another...");
                    uploadTasks.Remove(firstTaskFinished);
                }
            }

            // waiting for the remaining results to finish
            await Task.WhenAll(uploadTasks);

            DateTime endTime = DateTime.Now;

            TimeSpan runningTime = endTime - startTime;
            Console.WriteLine("\nEnded at: {0} \n", endTime);
            Console.WriteLine("Upload time total: {0}", runningTime);

            double timePerBatch = Math.Round(runningTime.TotalMilliseconds / (numDocs / batchSize), 4);
            Console.WriteLine("Upload time per batch: {0} ms", timePerBatch);

            double timePerDoc = Math.Round(runningTime.TotalMilliseconds / numDocs, 4);
            Console.WriteLine("Upload time per document: {0} ms \n", timePerDoc);
        }

このコードでは、バックオフ戦略を実装する関数 ExponentialBackoffAsync への非同期呼び出しを使用します。 プロセッサに含まれるコア数など、スレッドを使用して関数を呼び出します。 スレッドの最大数が使用されている場合、コードはスレッドが完了するまで待機します。 その後、すべてのドキュメントがアップロードされるまで、新しいスレッドが作成されます。