イベント処理アクションの作成#

#目的

この記事では、外部のデータを使用してData Processing Engineで定義されたアクションやワークフローをトリガーするためのアクションを作成する手順について説明します。

#イベント処理テーブル

アクションが外部イベントによって呼び出される場合、イベント呼び出しをプロジェクトの参照用にアーカイブすることをお勧めします。この記事では、データモデルにあるevent_handleというテーブルの例を使用します。

• テーブル名: event_handle
• テーブル説明: このテーブルには、イベントの内容を保存するための4つの属性が含まれています。

一般的に、イベントを処理するカスタムアクションを作成する予定の場合、これは良いベストプラクティスです。これは、すべてのトリガーされたイベントをアーカイブするために使用します。このようなテーブルを作成するには、Lakehouse Managerコンポーネントに移動し、上記の表に示されている属性に従ってテーブルを作成します。

Lakehouse Managerでデータモデルを作成する

#イベント処理アクションの設定

カスタムアクションを作成するには、Data Processing Engineコンポーネントに移動し、新しいカスタムアクションを作成し、アクション名をEvent handleと入力します。以下に示すサンプルPythonコードは、アクションがプラットフォーム外部（通常はAPIエンドポイント）でトリガーされた場合にイベントの本文を管理する方法を示しています。

カスタムアクションの設定で、関数名（メイン関数として実行するPython関数を指定する）をevent_handleに変更します。最後に、以下のコードサンプルをカスタムコードエディタにコピー＆ペーストします。

import logging
import pandas as pd
from datetime import datetime
from forepaas.dwh import bulk_insert
from forepaas.dwh import connect

logger = logging.getLogger(__name__)

def event_handle(event):
    logger.info(f"Start inserting event to default_dataset/event_handle")
    logger.info(f"event_type = {event.type}, number of rows = {len(event.content)}")
    if type(event.content) is not list:
        raise Exception("When playing this action you need to send rows")

    # Create dataframe
    df = pd.DataFrame(event.content)
    df["date"] = datetime.now().strftime("%Y-%m-%d")

    # Connect to destination database and insert dataframe to table
    destination = connect("dwh/default_dataset/")
    stats, error = bulk_insert(destination, "event_handle", df)

    logger.info(f"End insert of event: stats={stats}, error={error}")

#アクションとワークフローの起動

アクションが外部から起動された場合の受信イベント情報を処理する方法を知ったので、次にトリガーイベントを設定する方法を見てみましょう。これにより、アクションやワークフローのジョブをプログラムで開始できます。

#必須パラメータ

ジョブを起動するには3つのパラメータが必要です。これらは、当然ながら自分の環境構成に応じて設定する必要があります。すべてのパラメータは、{}括弧を使用してコードサンプルに示されています。

#プロジェクトサブドメイン

プロジェクトのサブドメインを取得するには、プロジェクトのURLを確認します。例えば、DPEコンポーネントを開いた状態でURLをコピー＆ペーストすると、次のようになります。

https://{project_subdomain}.eu.dataplatform.ovh.net/dpe/#/

サブドメインは、プロジェクトのURLのパスの最初の参照です。

#アクションIDまたはワークフローID

開始したいアクションのアクションIDを取得するには、アクションを編集する際のアクションのURLでも確認できます。そのためには、トリガーしたいアクションを編集するにはクリックし、URLは次のようになります。

# Action's URL
https://{project_subdomain}.eu.dataplatform.ovh.net/dpe/#/action/{action_id}

# Workflow's URL
https://{project_subdomain}.eu.dataplatform.ovh.net/dpe/#/workflow/{workflow_id}

#認証トークン

認証トークンを生成するには、Identity Access Managerコンポーネントから特定のAPI＆シークレットキーを作成する必要があります。その方法を学ぶには、次の専用記事を参照してください。

API / シークレットキーの作成

#コードサンプル

curl --request POST \
  --url 'https://{subdomain}.eu.dataplatform.ovh.net/dpe/v3/actions/{action_id}/start?app_id=55c1423560702d6426490f38&type=cam&token={token}' \
  --header 'content-type: application/json' \
  --data '[
	{
		"event_name": "Rebecca",
		"event_id": 1,
		"event_content": "imperdiet ornare."
	},
	{
		"event_name": "Veronica",
		"event_id": 2,
		"event_content": "Nullam ut nisi a odio"
	},
	{
		"event_name": "Aurelia",
		"event_id": 3,
		"event_content": "amet"
	},
	{
		"event_name": "Yardley",
		"event_id": 4,
		"event_content": "ornare placerat, orci lacus vestibulum"
	},
	{
		"event_name": "Dakota",
		"event_id": 5,
		"event_content": "nulla. Cras eu tellus eu"
	}
]'

import requests

url = "https://{subdomain}.eu.dataplatform.ovh.net/dpe/v3/actions/{action_id}/start"
querystring = {
  "app_id":"55c1423560702d6426490f38",
  "type":"cam",
  "token":"{token}"
}
payload = [
  {
    "event_name": "Rebecca",
    "event_id": 1,
    "event_content": "imperdiet ornare."
  },
  {
    "event_name": "Veronica",
    "event_id": 2,
    "event_content": "Nullam ut nisi a odio"
  },
  {
    "event_name": "Aurelia",
    "event_id": 3,
    "event_content": "amet"
  },
  {
    "event_name": "Yardley",
    "event_id": 4,
    "event_content": "ornare placerat, orci lacus vestibulum"
  },
  {
    "event_name": "Dakota",
    "event_id": 5,
    "event_content": "nulla. Cras eu tellus eu"
  }
]

response = requests.request("POST", url, params=querystring, json=payload) 
print(response.text)

この記事が役立ったことを願っています。質問があればお知らせください。改善案があればお知らせください。

#さらに詳しく

当社のソリューションを実装するためのトレーニングや技術サポートが必要な場合は、営業担当者にお問い合わせください。または、このリンクをクリックして、プロフェッショナルサービスの専門家にプロジェクトのカスタム分析を依頼する見積もりを取得してください。

Data Platformを構築しているチームと直接やり取りし、質問をする、フィードバックを提供する、専用のDiscordチャネルに参加してください。

OVHcloudサービスについてサポートが必要な場合は、ヘルプセンターでリクエストを作成してください。

コミュニティに参加してください。