Behavioral data push demo

Prerequisites

Before you begin, ensure that you have:

• An OpenSearch Industry Algorithm Edition application created in your Alibaba Cloud account

• A Resource Access Management (RAM) user with the required permissions. The AccessKey pair of an Alibaba Cloud account grants access to all API operations, so use a RAM user for API calls and routine O&M instead. See Create a RAM user to get started, then review Access authorization rules to grant the correct permissions

• The AccessKey ID and AccessKey secret of the RAM user. See Create an AccessKey pair

• (Optional) A Security Token Service (STS) token if you use RAM and STS authentication. Call the AssumeRole operation to get one

Set up environment variables

Set the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables before running the sample code.

Linux and macOS

Replace <access_key_id> and <access_key_secret> with the AccessKey ID and AccessKey secret of your RAM user.

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

Windows

1. Create an environment variable file and add ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET with their values.

2. Restart Windows for the changes to take effect.

Install dependencies

pip install alibabacloud_tea_util
pip install alibabacloud_opensearch_util
pip install alibabacloud_credentials

Sample code

The following code pushes behavioral events to OpenSearch using the behaviorBulk method, which sends a POST request to the bulk behavioral data endpoint.

# -*- 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__":

    # Specify the OpenSearch API endpoint.
    endpoint = "<endpoint>"

    # Specify the request protocol. Valid values: HTTPS and HTTP.
    endpoint_protocol = "HTTP"

    # Read credentials from environment variables.
    # Set the environment variables before running this code (see "Set up environment variables" above).
    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

    # Specify the authentication method.
    # Valid values: access_key (default), sts (RAM + STS authentication).
    auth_type = "sts"

    # Required only for STS authentication.
    # Call the AssumeRole operation of Alibaba Cloud RAM to get an STS token.
    security_token = "<security_token>"

    # Configure common request parameters.
    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)

    # Create an OpenSearch instance.
    ops = opensearch(Configs)
    app_name = "app_name"

    # --------------- Push behavior logs ---------------

    # item_id: the primary key ID returned in search results.
    item_id = "358713"

    # ops_request_misc: the request miscellaneous information returned in search results.
    ops_request_misc = "%7B%22request%5Fid%22%3A%22161777635816780357273903%22%2C%22scm%22%3A%2220140713.130149759..%22%7D"

    # bhv_type: the user action that triggered this event.
    # Valid values: expose, cart, collect, like, comment, buy, click
    bhv_type = "click"

    # request_id: the request ID returned in search results.
    request_id = "161777635816780357273903"

    # reach_time: when the server receives the event. Use a UNIX timestamp accurate to the second.
    reach_time = "1709708439"

    # user_id: the unique ID of the user who triggered the event.
    # Use the logged-in user ID; use the cookie ID for unauthenticated users on PC clients.
    user_id = "a7a0d37c824b659f36a5b9e3b819fcdd"
    behavior_fields1 = behavior_fields2 = {
        "item_id": item_id,
        "sdk_type": "opensearch_sdk",
        "sdk_version": "<sdk_version>", # Current OpenSearch SDK for Python version: 3.2.0
        {"trace_id", "ALIBABA"}, # Service provider.
        "trace_info": ops_request_misc,
        "bhv_type": bhv_type,
        "item_type": "item",
        "rn": request_id,
        "biz_id":"<biz_id>", # Numerical ID for differentiating business across mobile apps or clients. Can be linked to OpenSearch applications and Artificial Intelligence Recommendation (AIRec) instances.
        "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)

Behavioral event types

Use the bhv_type field to specify what action the user performed.

Behavioral data fields

Each behavioral event is a JSON object in the fields key of a document with "cmd": "add". The table below lists all fields and their requirements.

What's next

• Push collected data

• AliyunServiceRoleForOpenSearch