非同期推論、キューサービス - Platform For AI - Alibaba Cloud ドキュメントセンター

PAI EAS は、非同期推論キューと連携するための 3 つのインターフェイス (HTTP API、Python SDK、eascmd CLI) を提供しています。このドキュメントでは、これら 3 つすべてを説明します。

仕組み

非同期推論サービスをデプロイすると、EAS は自動的に 2 つのキューを作成します：

入力キュー：クライアントからの推論リクエストを受け付けます。
出力キュー (シンク)：推論サービスが書き込んだ推論結果を格納します。

データフローは次のとおりです：

入力キューに推論リクエストを送信します。EAS は直ちにリクエスト ID とキューインデックスを返します。
推論サービスが入力キューからデータを読み取り、推論を実行し、結果を出力キューに書き込みます。
リクエスト ID またはインデックスを使用して、出力キューから結果を取得します。

前提条件

開始する前に、次の項目が必要です。

PAI EAS にデプロイ済みの非同期推論サービス
サービスの入力キューのエンドポイント、出力キューのエンドポイント、およびトークン

これらの値を確認するには、[推論サービス] タブに移動し、サービス名をクリックして [概要] ページを開き、[基本情報] セクションにある [エンドポイント情報を表示] をクリックします。[エンドポイント情報を表示] ダイアログボックスで、[共有ゲートウェイ] > [非同期呼び出し] タブをクリックします。エンドポイントは、ネットワークタイプ ([インターネット] および [VPC]) 別にグループ化されており、各グループにはキューサービスの入力エンドポイントと出力エンドポイントが用意されています。

エンドポイントの形式は次のとおりです：

エンドポイント	形式	例
入力キュー	`{domain}/api/predict/{service_name}`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}`
出力キュー (シンク)	`{domain}/api/predict/{service_name}/sink`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}/sink`

APIによるキューサービスへのアクセス

すべての HTTP リクエストには、サービスのトークンを含む Authorization ヘッダーが必要です。

クイックリファレンス：

操作	メソッド	パス	主要パラメーター
データ送信	POST	`/api/predict/{service}`	`_priority_=1` (任意)
キューの詳細表示	GET	`/api/predict/{service}?_attrs_=true`	—
インデックスによるデータ照会	GET	`/api/predict/{service}?_index_=N`	`_length_`、`_auto_delete_`、`_timeout_`
リクエスト ID によるデータ照会	GET	`/api/predict/{service}?requestId=...`	`_timeout_`
推論結果の照会	GET	`/api/predict/{service}/sink?requestId=...`	`_timeout_`
単一アイテムの削除	DELETE	`/api/predict/{service}?_index_=N`	—
キューの切り捨て	DELETE	`/api/predict/{service}?_index_=N&_trunc_=true`	—

キューへのデータ送信

curl を使用して、入力キューにリクエストを送信します。

curl -v http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice \
  -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  -d '[{}]'

レスポンスには次の内容が含まれます：

> POST /api/predict/qservice HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< X-Eas-Queueservice-Request-Id: 4e034bnvb-e783-4272-9333-68x6a1v8dc6x
<
1033

後でデータを照会する際に使用できる、2 つの識別子が返されます：

X-Eas-Queueservice-Request-Id ヘッダー：リクエスト ID (4e034bnvb-e783-4272-9333-68x6a1v8dc6x) です。
レスポンスボディ：キューインデックス (1033) です。

優先データの送信

キューは、デフォルトでは先入れ先出し (FIFO) 順でデータを処理します。特定のリクエストを優先させるには、クエリ文字列に _priority_=1 を追加します。

curl -v "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_priority_=1" \
  -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  -d '[{}]'

優先データは、通常のデータより先にサブスクライバーにプッシュされます。

キューの詳細表示

キューメタデータを取得するには、GET リクエストに _attrs_=true を追加します。

curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_attrs_=true"

レスポンスは JSON オブジェクトです：

{"consumers.stats.total":"0","consumers.status.total":"0","meta.header.group":"X-EAS-QueueService-Gid","meta.header.priority":"X-EAS-QueueService-Priority","meta.header.user":"X-EAS-QueueService-Uid","stream.maxPayloadBytes":"524288","meta.name":"pmml_test","meta.state":"Normal","stream.approxMaxLength":"230399","stream.firstEntry":"0","stream.lastEntry":"0","stream.length":"1"}

主要フィールド：

フィールド	説明
`stream.maxPayloadBytes`	単一データアイテムの最大サイズ (バイト)
`stream.approxMaxLength`	キューが保持できるデータアイテムの最大数
`stream.firstEntry`	キュー内の先頭アイテムのインデックス
`stream.lastEntry`	キュー内の末尾アイテムのインデックス
`stream.length`	キュー内の現在のアイテム数
`meta.state`	現在のキュー状態

または、[Elastic Algorithm Service (EAS)] ページに移動し、サービス名をクリックして、[非同期キュー] タブに切り替えます。

このタブには、キューの基本情報 (所属するリソースグループ、作成時刻、単一入力リクエストの最大サイズ、単一出力レスポンスの最大サイズ)、サービス向けにデプロイされているリソース (インスタンス数、CPU、メモリ)、および入力キューに格納されている現在のデータアイテム数と各インスタンスの処理ステータスが表示されます。

データ照会

ユースケースに応じて取得方法を選択します：

方法	仕組み	利用シーン
ポーリング	必要に応じて、インデックスまたはリクエスト ID でデータを照会します。	特定アイテムの取得、またはキュー状態の断続的な確認
サブスクリプション	WebSocket 経由でサブスクライブし、到着するデータを受信します。	継続的なストリーム処理、または高スループットのシナリオ

入力キューからのポーリング

インデックスまたはリクエスト ID でデータを照会します：

# インデックスで照会
curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022"

# リクエスト ID で照会
curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?requestId=87633037-39a4-40bf-8405-14f8e0c31896"

レスポンス：

> GET /api/predict/qservice?_index_=1022&_auto_delete_=false HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
[{}]

レスポンスステータスコード：

ステータスコード	意味
`200 OK`	データが見つかり、返されました。
`204 No Content`	一致するデータが存在しません (`_timeout_=0` の場合は即座に返されます)。

クエリパラメーター：

パラメーター	型	デフォルト	説明
`_index_`	INT	`0`	開始インデックスです。照会効率を高めるため、対象アイテムのインデックスに近い値を設定してください。
`_length_`	INT	`1`	取得するアイテム数です。
`_auto_delete_`	BOOL	`true`	取得後に、照会したアイテムをキューから削除します。
`_timeout_`	STRING	`0`	一致するデータが存在しない場合の待機時間です。`0` を指定すると、すぐに HTTP 204 が返されます。値の例: `1s`、`1m`。
`requestId`	STRING	—	特定のアイテムを特定するために使用される組み込みタグです。EAS フレームワークは、各入力アイテムに `requestId` でタグ付けし、それを出力キューまで引き継ぎます。これにより、同じリクエスト ID を使用して出力キューから推論結果を照会できます。

出力キューからの推論結果のポーリング

元の送信で取得したリクエスト ID を使用して、出力キューから結果を照会します：

curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12bbe2d"

レスポンス：

> GET /api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12b**** HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 53
< Content-Type: text/plain; charset=utf-8
<
[{"p_0":0.5224580736905329,"p_1":0.4775419263094671}]

データ削除

キューからデータを削除する方法は 2 つあります。単一アイテムの削除、または指定したインデックスまでのすべてのアイテムの切り捨てです。

単一アイテムの削除

curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022"

レスポンス：

> DELETE /api/predict/qservice?_index_=1022 HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
OK

パラメーター：

パラメーター	型	説明
`_index_`	INT	削除するアイテムのインデックス

キューの切り捨て

指定した値より小さいインデックスを持つすべてのアイテムを削除します：

curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1023&_trunc_=true"

レスポンス：

> DELETE /api/predict/qservice?_index_=1023&_trunc_=true HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
OK

パラメーター：

パラメーター	型	説明
`_index_`	INT	カットオフインデックスです。この値より小さいインデックスを持つすべてのアイテムが削除されます。
`_trunc_`	BOOL	切り捨てを実行するには`true`にする必要があり、省略した場合や`false`の場合は、代わりに単一アイテムの削除が実行されます。

キューのサブスクライブ (Python SDK)

継続的なストリーム処理を行う場合は、Python SDK を使用して出力キューをサブスクライブします。キューサービスは WebSocket プロトコルを使用して永続的な接続を維持し、データが到着するとサブスクライバーにプッシュします。サブスクリプションのウィンドウサイズは、推論サービスインスタンスの worker_threads 設定で制御されます。

推論サービスは必須ではありません。SDK を使用してカスタムサービス内で入力キューをサブスクライブし、結果をサードパーティのメッセージキューや Object Storage Service (OSS) などのストレージに書き込むこともできます。

SDK のインストール：

pip install eas_prediction --user

出力キューのサブスクライブ：

次の例では、QueueClient を使用して入力キューに 10 個のアイテムを送信し、出力キューからの結果をサブスクライブします。本番環境では、送信とサブスクライブには別々のスレッドを使用してください。

#!/usr/bin/env python
from eas_prediction import QueueClient

# 入力キュー用のクライアントを作成します。
input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice')
# カスタムユーザーまたはグループを設定する場合：
# input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice', uid='your_user_id', gid='your_group_id')
input_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==')
input_queue.init()

# 出力キュー用のクライアントを作成します。
sink_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice/sink')
sink_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==')
sink_queue.init()

# 入力キューに 10 個のアイテムを送信します。
for x in range(10):
    index, request_id = input_queue.put('[{}]')
    print(index, request_id)

    # キュー属性を出力します。
    attrs = input_queue.attributes()
    print(attrs)

# ウィンドウサイズ 5 で出力キューをサブスクライブします。
i = 0
watcher = sink_queue.watch(0, 5, auto_commit=False)
for x in watcher.run():
    print(x.data.decode('utf-8'))

    # 処理後に各アイテムをコミットします。
    sink_queue.commit(x.index)
    i += 1
    if i == 10:
        break

# ウォッチャーを閉じます。各 QueueClient インスタンスがサポートするアクティブなウォッチャーは、同時に1つだけです。
# ウォッチャーを閉じないと、次の watch() 呼び出しでエラーが発生します。
watcher.close()

次のステップ

Queue service subscription and push：コンシューマーグループ、ユーザー、コミットのセマンティクスについて説明します。
クライアントのダウンロードと認証ー eascmd のセットアップと更新。