このトピックでは、API 操作を呼び出すか、SDK または EASCMD クライアントを使用して、キューサービスにアクセスする方法について説明します。
API 操作を呼び出してキューサービスにアクセスする
非同期推論サービスをデプロイした後、入力キューと出力キュー（シンクキュー）に対して 2 種類のアドレスが自動的に生成されます。次の表に、HTTP エンドポイントの例を示します。
エンドポイントタイプ | エンドポイント形式 | 例 |
入力キューエンドポイント |
|
|
出力キューエンドポイント |
|
|
[Elastic Algorithm Service (EAS)El] タブで、対象のサービス名をクリックして [概要] ページに移動します。[基本情報] セクションで [エンドポイント情報の表示] をクリックして、入力キューアドレス、出力キューアドレス、およびトークンを取得します。
キューサービスにデータを送信する
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: 19
< X-Eas-Queueservice-Request-Id: 4e034bnvb-e783-4272-9333-68x6a1v8dc6x
<
1033上記の応答では：
応答ヘッダーの X-Eas-Queueservice-Request-Id の値 4e034bnvb-e783-4272-9333-68x6a1v8dc6x はリクエスト ID です。リクエスト ID を使用してデータをクエリできます。
応答本文の Index の値 1033 は、キュー内のリクエストのインデックスです。インデックスを使用して、キュー内のデータをクエリできます。
優先データの送信
データは、先入れ先出し（FIFO）の順序でキューサービスにプッシュされます。ただし、多くのシナリオでは、特定のデータを最初にプッシュして処理する必要があります。キューサービスは、データの優先順位に基づいてデータをプッシュすることをサポートしています。 _priority_=1 パラメーターを追加して、優先順位の高いデータをキューサービスにプッシュできます。
$ curl -v http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_priority_=1 -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' -d '[{}]'次の出力が返されます。
> POST /api/predict/qservice?_priority_=1 HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 19
< X-Eas-Queueservice-Request-Id: 4033eb55-e783-4922-9777-68d6a1383c76
<
1034優先データがキューに書き込まれた後、データは優先順位としてサブスクライバーにプッシュされます。
キューサービスの詳細を表示する
キューサービスにリクエストを送信するときに _attrs_=true パラメーターを追加すると、キューの詳細はレスポンスで返されます。サンプルコード：
$ curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_attrs_=true次の出力が返されます。
> GET /api/predict/qservice?_attrs_=true HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 320
<
{"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":"4095","stream.firstEntry":"0","stream.lastEntry":"0","stream.length":"1"}返されるコンテンツは JSON 形式です。次の表に、フィールドを示します。
フィールド | 説明 |
stream.maxPayloadBytes | キュー内の各データエントリの最大サイズ。単位：バイト。 |
stream.approxMaxLength | キューに格納できるデータエントリの最大数。 |
stream.firstEntry | キュー内の最初のデータエントリのインデックス。 |
stream.lastEntry | キュー内の最後のデータエントリのインデックス。 |
stream.length | キューに格納されているデータエントリの数。 |
meta.state | キューのステータス。 |
また、Elastic Algorithm Service (EAS) ページで非同期推論サービスの名前をクリックして [概要] ページに移動し、[非同期キュー] タブに切り替えてキュー情報を表示することもできます。
データをクエリする
条件に基づいてデータをクエリする
1 つのキューサービスのみを使用する場合は、インデックスまたはリクエスト 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 < [{}]必要なパラメーターを構成して、推論結果をクエリできます。次の表に、パラメーターを示します。
パラメーター
タイプ
説明
_index_
INT
クエリするデータの開始インデックス。デフォルト値は 0 で、最初のデータエントリからクエリを開始することを指定します。インデックスがクエリ対象のデータに近いほど、クエリの効率が高くなります。
_length_
INT
クエリするデータエントリの数。デフォルト値は 1 で、1 つのデータエントリのみがクエリされることを指定します。
_auto_delete_
BOOL
クエリされたデータエントリをキューから削除するかどうかを指定します。デフォルト値：TRUE。この値は、クエリが完了した後、クエリされたデータエントリがキューから自動的に削除されることを指定します。
_timeout_
STRING
タイムアウト期間。デフォルト値は 0 で、クエリ条件に一致するデータがない場合に 204 状態コードがすぐに返されることを指定します。それ以外の場合、リクエストは指定された時間待機します。タイムアウト期間内にクエリ条件に一致するデータがキューに存在する場合、データが返されます。例：1s（1 秒）、1m（1 分）。
requestId
STRING
requestId は、クエリ条件として使用できる組み込みタグです。
説明推論サービスを非同期に呼び出すと、入力キューはリクエストを返します。Elastic Algorithm Service (EAS) サービスフレームワークは、データを取得、処理し、出力キューに書き込みます。サービスフレームワークは、requestId タグを使用して入力データと出力データを関連付けます。入力データのリクエスト ID を使用して、出力キューから推論結果を取得できます。
非同期推論結果をクエリする
キューサービスを推論サービスと一緒に使用すると、推論サービスは入力キューからリクエストデータを自動的に取得し、推論を実行して、結果を出力キュー（シンクキュー）に書き込みます。次のサンプルコードは、リクエスト ID 0337f7a1-a6f6-49a6-8ad7-ff2fd12b**** を使用して出力キューサービスからデータをクエリする方法の例を示しています。
$ 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}]
データをクリーンアップする
キュー内の特定のデータが不要になった場合は、API 操作を呼び出すことによってデータをクリーンアップできます。単一のデータエントリを削除するか、データの切り捨てを実行して複数のデータエントリを削除することで、データをクリアできます。
単一のデータエントリを削除する
# インデックスを使用してデータを削除します。 $ 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 に設定します。それ以外の場合、単一のデータエントリのみが削除されます。
キューサービスにサブスクライブする
クエリに加えて、キューサービスにサブスクライブして非同期推論結果を取得できます。キューサービスは、サブスクリプションインターフェイスとして watch を提供します。クライアントは watch インターフェイスを使用して推論結果を取得できます。キューサービスは、推論サービスインスタンスの上限同時実行数（worker_threads）に基づいてサブスクリプションウィンドウサイズを制御します。新しいデータがキューに書き込まれると、キューサービスは自動的にデータをクライアントにプッシュします。
この機能は、WebSocket ベースの SDK にカプセル化された QueueClient を使用して実装されます。この機能は、永続的な接続を介してデータをプッシュします。次の例では、ビデオおよびオーディオストリーム処理サービスを使用して、Python 用 EAS SDK の QueueClient を使用してキュー内のデータにサブスクライブする方法について説明します。
推論サービスはオプションです。SDK を使用してカスタムサービスの入力キューにサブスクライブし、出力データをサードパーティのメッセージキューまたはデータストアに書き込むことができます。たとえば、イメージを Object Storage Service (OSS) に出力できます。
Python 用 EAS SDK をインストールします。
pip install eas_prediction --userQueueClient の
put() 関数を使用して入力キューにデータを送信し、watch() 関数を使用して出力キューのデータにサブスクライブできます。データの送信とサブスクリプションは、異なるスレッドで発生する可能性があります。この例では、両方とも put 関数と watch 関数を使用して同じスレッドで完了します。#!/usr/bin/env python from eas_prediction import QueueClient # 入力データを受信するための入力キューオブジェクトを作成します。 input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice') # カスタムユーザーとグループを作成するには、uid を使用してユーザーを、gid を使用してグループを指定します。例： # 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 # watcher オブジェクトを無効にします。各クライアントは 1 つの watcher オブジェクトのみを使用できます。watcher オブジェクトを無効にしないと、クライアントを再実行するとエラーが発生します。 watcher.close()
EASCMD クライアントを使用してキューサービスにアクセスする
EASCMD クライアントは、完全なキューサービス API をカプセル化します。 eascmd stream サブコマンドを実行して、キューサービスを管理およびデバッグできます。
EASCMD クライアントをダウンロードする
EASCMD のバージョンが 2.6.0 以降であることを確認してください。EASCMD クライアントのダウンロード、更新、および構成方法については、「EASCMD クライアントをダウンロードしてユーザー認証を完了する」をご参照ください。
EASCMD を構成してキューサービスにアクセスする
次の eascmd stream config コマンドを実行して、キューサービスへのアクセスを構成します。
eascmd stream config --url=http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice --token=YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==構成が完了すると、EASCMD は default_group をデフォルトの group_id として、default_user をデフォルトの user_id として使用します。グループとユーザーについては、「キューサービスのサブスクリプションとプッシュ」をご参照ください。 -- group パラメーターと -- user パラメーターを使用して、カスタム group_id と user_id を作成できます。 eascmd stream config コマンドのすべてのパラメーターは、他の読み取りおよび書き込みコマンドによって上書きできます。
キューの詳細をクエリする
info コマンドを実行して、キューの詳細を表示します。サンプルコード：
eascmd stream info次の出力が返されます。
[OK] Attributes:
consumers.list.[0] : Id: imageasync.imageasync-35d72370-5f576f7c8d-2mdb4, Index: 0, Pending: 0, Status: Running, Idle: 19.997s, Window: 5, Slots: 5, AutoCommit: false
consumers.stats.total : 1
consumers.status.total : 1
groups.list.[0] : Id: imageasync, Index: 0, Pending: 0, Delivered: 1, Consumers: 1
meta.header.group : X-EAS-QueueService-Gid
meta.header.priority : X-EAS-QueueService-Priority
meta.header.user : X-EAS-QueueService-Uid
meta.maxPayloadBytes : 8192
meta.name : imageasync-queue-38895e88
meta.state : Normal
stream.approxMaxLength : 230399
stream.firstEntry : 0
stream.lastEntry : 0
stream.length : 0レスポンスのパラメーターについては、このトピックの「API 操作を呼び出してキューサービスにアクセスする」セクションをご参照ください。 info コマンドを使用して、キューの属性を取得し、キューサービスへの接続性をテストできます。
キューにデータを送信する
put コマンドを実行して、キューにデータを送信します。サンプルコード：
eascmd stream put -d "10s"次の出力が返されます。
[OK] 1
[INFO] Put data done.
Total time cost: 401.892141ms
Total size: 3.00 B
Total: 1, success: 1, failed: 0-f パラメーターを使用して、ファイル内のすべてのデータをキューに送信することもできます。サンプルコード：
eascmd stream put -f test.data次の出力が返されます。
[INFO] Opening data file: test.data
[OK] 2
[OK] 3
[OK] 4
[OK] 5
[OK] 6
[OK] 7
[OK] 8
[OK] 9
[OK] 10
[OK] 11
[OK] 12
[OK] 13
....この場合、info コマンドを実行してキューの状態を表示できます。
キュー内のデータをクエリする
get コマンドを実行して、キューからデータをクエリします。サンプルコード：
eascmd stream get -l10 --timeout=3s次の出力が返されます。
[OK] [0 - 1] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=e47b76e2-2648-40fe-9197-a268015cbd1f ts@source=1685802680575] data1
[OK] [1 - 2] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=51d13952-6ba3-4d52-b548-e58837675c7a ts@source=1685807531686] data2
[OK] [2 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OK
[OK] [3 - 4] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=2a5645b6-9ee5-4026-bdee-fab31e435934 ts@source=1685807531715] data4
[OK] [4 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@source=1685807531730] data5
...受信した最初のデータエントリでは、次のとおりです。
[0 - 1] 列は、データ 0 のインデックスが 1 であることを示します。
tags[Header:Content-Type=text/plain; charset= ...]列は、データにタグが含まれていることを示します。上記のレスポンスでは、次のとおりです。Header は、データを入力したときの HTTP リクエストのヘッダーを示します。
requestId は、自動的に生成される組み込みリクエスト ID を示します。
ts@source は、入力キューがリクエストを受信したときの UNIX タイムスタンプを示します。 ts@sink は、出力キューがデータを受信したときのタイムスタンプを示します。
最後の列は、入力したデータを示します。
キューが推論サービスインスタンスと一致する場合、データがキューに送信された後、入力データがインスタンスによって消費される可能性があります。この場合、-k パラメーターをコマンドに追加して、出力キューのデータをクエリします。
-- tags パラメーターを使用してクエリ条件を追加することもできます。たとえば、次のコマンドを実行して、requestId を使用してデータをクエリします。
eascmd stream get --tags requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f次の出力が返されます。
[OK] [0 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OKキューからデータを削除する
delete コマンドを実行して単一のデータエントリを削除するか、trunc コマンドを実行して複数のデータエントリを削除できます。次のセクションでは例を示します。
単一のデータエントリを削除する：
eascmd stream delete 3データエントリが削除されると、次の出力が返されます。
Deleting index(es):
3 [y/N]y
[OK] deleted複数のデータエントリを削除する：
eascmd stream trunc 4データエントリが削除されると、次の出力が返されます。
trunc stream from index: 4 [y/N]y
[OK] truncatedキューにサブスクライブする
watch コマンドを実行して、キューサービスにサブスクライブします。サンプルコード：
eascmd stream watch次の出力が返されます。
[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: false, window: 10
I0604 09:20:45.211243 66197 queue.go:532] watch via websocket
[OK] [0 - 4] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=2a5645b6-9ee5-4026-bdee-fab31e435934 ts@sink=1685807531718 ts@source=1685807531715] data4
commit: 4 ? [Y/n]Y と入力してコミットし、新しいデータを取得します。
[OK] [1 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
commit: 5 ? [Y/n]n と入力して、ネガティブコミットを実行するかどうかを指定します。
[OK] [1 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
commit: 5 ? [Y/n]n
negative: 5 ? [Y/n]yコミットとネガティブコミットについては、「キューサービスのサブスクリプションとプッシュ」トピックの「コミットとネガティブ」セクションをご参照ください。
-- auto-commit オプションを使用すると、サーバーでデータが自動的にコミットされます。
eascmd stream watch --auto-commit次の出力が返されます。
[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: true, window: 10
I0604 09:30:08.554542 66408 queue.go:532] watch via websocket
[OK] [0 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
[OK] [1 - 6] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=5825dd3e-a5e2-4754-a946-96e068d643c8 ts@sink=1685807531771 ts@source=1685807531768] data6
[OK] [2 - 7] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=e7edf9b8-de78-41a0-8d9c-0a4aaf7dcaaf ts@sink=1685807531786 ts@source=1685807531783] data7
[OK] [3 - 8] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=3ddc3481-934a-4408-8d08-11c2c2248ef6 ts@sink=1685807531801 ts@source=1685807531798] data8
[OK] [4 - 9] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=561da95d-b99a-4710-bb82-9402baa21f36 ts@sink=1685807531816 ts@source=1685807531812] data9
....その他のオプションとコマンド
前のセクションでは、eascmd stream の最も一般的に使用されるコマンドとオプションについて説明しました。 eascmd stream の拡張機能については、eascmd stream help コマンドを実行してください。