OpenSearch:行動データのプッシュデモ

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

• Alibaba Cloud アカウントで作成済みの OpenSearch Industry Algorithm Edition アプリケーション

• 必要な権限を持つ Resource Access Management (RAM) ユーザー。Alibaba Cloud アカウントの AccessKey ペアはすべての API オペレーションへのアクセスを許可するため、API 呼び出しおよび通常の O&M には RAM ユーザーを使用してください。開始するには、「RAM ユーザーの作成」をご参照ください。その後、「アクセス権限付与ルール」を確認して、適切な権限を付与してください。

• RAM ユーザーの AccessKey ID と AccessKey Secret。「AccessKey ペアを作成する」をご参照ください。

• （任意）RAM および STS 認証を利用する場合の Security Token Service (STS) トークン。AssumeRole オペレーションを呼び出して取得します。

環境変数の設定

サンプルコードを実行する前に、ALIBABA_CLOUD_ACCESS_KEY_ID および ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数を設定してください。

Linux および macOS

<access_key_id> および <access_key_secret> を、ご利用の RAM ユーザーの AccessKey ID および AccessKey Secret に置き換えてください。

export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>

Windows

1. 環境変数ファイルを作成し、ALIBABA_CLOUD_ACCESS_KEY_ID および ALIBABA_CLOUD_ACCESS_KEY_SECRET をそれぞれの値とともに追加します。

2. 変更を有効にするために Windows を再起動します。

依存関係のインストール

pip install alibabacloud_tea_util
pip install alibabacloud_opensearch_util
pip install alibabacloud_credentials

サンプルコード

以下のコードでは、behaviorBulk メソッドを使用して行動イベントを OpenSearch にプッシュします。このメソッドは、バルク行動データエンドポイントに対して POST リクエストを送信します。

# -*- coding: utf-8 -*-

import time, os
from typing import Dict, Any

from Tea.exceptions import TeaException
from Tea.request import TeaRequest
from alibabacloud_tea_util import models as util_models

from BaseRequest import Config, Client


class opensearch:
    def __init__(self, config: Config):
        self.Clients = Client(config=config)
        self.runtime = util_models.RuntimeOptions(
            connect_timeout=10000,
            read_timeout=10000,
            autoretry=False,
            ignore_ssl=False,
            max_idle_conns=50,
            max_attempts=3
        )
        self.header = {}

    def behaviorBulk(self, app_name: str, collections_name: str, doc_content: list) -> Dict[str, Any]:
        try:
            response = self.Clients._request(method="POST",
                                             pathname=f'/v3/openapi/app-groups/{app_name}/data-collections/{collections_name}/data-collection-type/BEHAVIOR/actions/bulk',query={},headers = self.header,
                                             body=doc_content, runtime=self.runtime)
            return response
        except Exception as e:
            print(e)

if __name__ == "__main__":

    # OpenSearch API エンドポイントを指定します。
    endpoint = "<endpoint>"

    # リクエストプロトコルを指定します。有効な値：HTTPS、HTTP。
    endpoint_protocol = "HTTP"

    # 環境変数から認証情報を読み込みます。
    # このコードを実行する前に、環境変数を設定してください（上記「環境変数の設定」を参照）。
    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

    # 認証方式を指定します。
    # 有効な値：access_key（デフォルト）、sts（RAM ＋ STS 認証）。
    auth_type = "sts"

    # STS 認証を利用する場合のみ必須です。
    # Alibaba Cloud RAM の AssumeRole オペレーションを呼び出して STS トークンを取得します。
    security_token = "<security_token>"

    # 共通リクエストパラメーターを設定します。
    Configs = Config(endpoint=endpoint, access_key_id=access_key_id, access_key_secret=access_key_secret,
                     security_token=security_token, type=auth_type, protocol=endpoint_protocol)

    # OpenSearch インスタンスを作成します。
    ops = opensearch(Configs)
    app_name = "app_name"

    # --------------- 行動ログのプッシュ ---------------

    # item_id：検索結果で返されるプライマリキー ID。
    item_id = "358713"

    # ops_request_misc：検索結果で返されるリクエスト付随情報。
    ops_request_misc = "%7B%22request%5Fid%22%3A%22161777635816780357273903%22%2C%22scm%22%3A%2220140713.130149759..%22%7D"

    # bhv_type：このイベントをトリガーしたユーザー操作。
    # 有効な値：expose、cart、collect、like、comment、buy、click
    bhv_type = "click"

    # request_id：検索結果で返されるリクエスト ID。
    request_id = "161777635816780357273903"

    # reach_time：サーバーがイベントを受信した時刻。秒単位の UNIX タイムスタンプを使用します。
    reach_time = "1709708439"

    # user_id：このイベントをトリガーしたユーザーの固有 ID。
    # 認証済みユーザーにはログインユーザー ID を使用し、PC クライアントの未認証ユーザーには Cookie ID を使用します。
    user_id = "a7a0d37c824b659f36a5b9e3b819fcdd"
    behavior_fields1 = behavior_fields2 = {
        "item_id": item_id,
        "sdk_type": "opensearch_sdk",
        "sdk_version": "<sdk_version>", # 現在の OpenSearch Python SDK のバージョン：3.2.0
        {"trace_id", "ALIBABA"}, # サービスプロバイダー。
        "trace_info": ops_request_misc,
        "bhv_type": bhv_type,
        "item_type": "item",
        "rn": request_id,
        "biz_id":"<biz_id>", # モバイルアプリまたはクライアント間でビジネスを区別するための数値 ID。OpenSearch アプリケーションおよび Artificial Intelligence Recommendation (AIRec) インスタンスと関連付け可能です。
        "reach_time": reach_time,
        "user_id": user_id,
    }

    behavior_documents = [{"cmd": "add", "fields": behavior_fields1}, {"cmd": "add", "fields": behavior_fields2}]
    res6 = ops.behaviorBulk(app_name=app_name, collections_name=app_name, doc_content=behavior_documents)
    print(res6)

行動イベントタイプ

bhv_type フィールドを使用して、ユーザーが実行した操作を指定します。

行動データフィールド

各行動イベントは、fields キーを持つ JSON オブジェクトであり、そのドキュメントには "cmd": "add" が含まれます。下表に、すべてのフィールドとその要件を示します。

次のステップ

• 収集したデータをプッシュする

• OpenSearch用のAliyunサービスロール