はじめに
この記事はIBM Watson Studio 内のベータ版機能Pipelineで作成したフローをJobとして登録し、スケジュール実行と実行環境からのAPI呼び出し2種類の方法で動かす方法をご紹介します。
前提
- IBM Watson StudioのPipelineによって実行可能なパイプラインフローを構築済みであること
- パイプラインフローの構築手順についてはこちらの記事をご参照ください。
- （実行環境からのAPI呼び出しを実行する場合）ローカルPCにてnotebookを実行できること
#1. Jobをスケジュール実行
1つ目の実行手段として、Jobのスケジュール実行手順をご紹介いたします。
##スケジュール付きJobの登録
-
パイプラインが完成したら、キャンバス左上の「実行」から「ジョブの作成」を選択します。
-
「詳細の定義」、「API鍵」を任意の設定で入力し「スケジュール」まで進みます。
-
「スケジュール」にて「実行をスケジュール」を押下し、作成するJobのスケジュール設定を行います。
スケジュール設定では「開始」にてスケジュールの開始日時を、「繰り返し」にてスケジュールの実行間隔、除外日および終了日時を設定可能です。
設定が完了したら「次へ」を押下します。
-
「レビューして作成」まで進んだら、作成したスケジュール設定の内容を確認し「作成」を押下します。
##結果確認
プロジェクトのメイン画面から「ジョブ」タブを押下し、ジョブ一覧から作成したジョブを選択します。
ジョブの詳細画面からスケジュールの実行履歴を確認することができます。
#2. JobをAPI実行
2つ目の実行手段として、JobをAPI呼び出しにて実行する手順をご紹介します。
今回は実行環境としてローカルのJupyter notebookを想定しております。
##Jobの作成
- スケジュール実行時と同じく、キャンバス左上の「実行」から「ジョブの作成」を選択します。
- 今度はスケジュールの設定を行わずに、任意の設定でJobの作成を完了させます。
認証
APIの設定
IBM Cloudダッシュボード画面からIBM Cloud APIの作成をします。（APIキーを既に作成しており、値がわかっている場合はAPIキーの新規作成は不要です。）
- IBM Cloudダッシュボート右上の「管理」から「アクセス（IAM）」を選択し、左側メニューより「APIキー」を開きます。APIキー一覧画面の上の「IBM Cloud APIキーの作成」を押下します。
2.適当なAPIキーの管理名を入力し、「作成」ボタンを押下します。
3.APIキーが新規に作成されたら、「コピー」または「ダウンロード」からAPIキーを保存します。
IBM Cloud APIキーが作成できたらnotebookにAPIキーを格納します。
# APIKeyの設定
apikey = "<APIキー>"
トークンの取得
先ほど取得したAPIキーを使用しaccessTokenを取得、格納します。
import requests
# トークンの取得
def get_token(apikey):
url = "https://iam.ng.bluemix.net/identity/token"
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
data = "apikey=" + apikey + "&grant_type=urn:ibm:params:oauth:grant-type:apikey"
token = requests.post(url, headers=headers, data=data).json()['access_token']
return token
token = get_token(apikey)
##プロジェクト一覧を取得
以降はWatson Data APIを使用して操作をしていきます。
まず、Pipelineから作成したjobを配置しているプロジェクトのIDを調べるために一覧を表示させます。
使用APIはGET /v2/projectsです。
# 共通URLの設定
url = 'https://api.dataplatform.cloud.ibm.com'
# 共通ヘッダーの設定
headers = {
'Authorization': 'Bearer ' + token,
'Content-Type': 'application/json'
}
# プロジェクト一覧の出力
response_projects = requests.get(url+"/v2/projects", headers=headers, verify=False).json()
response_projects
# -output-
# {'total_results': 8,
# 'resources': [{'metadata': {'guid': '<プロジェクトID>',
# 'url': '/v2/projects/<プロジェクトID>',
# 'created_at': '2020-05-12T08:04:30.419Z',
# 'updated_at': '2020-05-12T08:10:01.773Z'},
# 'entity': {'name': 'projectName',
出力された一覧から目的のプロジェクトのIDを確認します(該当プロジェクトmetadataキー内の'guid'の値がプロジェクトIDになります)。
##Job一覧を取得
前ステップで取得したプロジェクトIDを格納し、Job一覧を表示させます。
使用APIはGET /v2/jobsです。
# 対象projectIDの格納
project_id = "<プロジェクトID>"
# 対象project内のjob一覧
response_jobs = requests.get(url+"/v2/jobs?project_id="+project_id, headers=headers, verify=False).json()
response_jobs
# -output-
# {'total_rows': 4,
# 'results': [{'metadata': {'name': 'jobName',
# 'description': 'It's test job',
# 'asset_id': '<job ID>',
目的のJobの'asset_id'が目的のJob IDとなります。
##Jobを実行
前ステップで取得したJob IDを格納し、Jobを実行します。
また作成したフローにパイプラインパラメーターを設定している場合は、実行の前にJob実行時のパイプラインパラメーターの設定を行います。
使用APIはPOST /v2/jobs/{job_id}/runsです。
# Job IDの格納
job_id='<Job ID>'
# 実行時のパイプラインパラメーター
jobrunpost = {
"job_run": {
"configuration" : {
'parameter_set': {'parameters': [
{'name': 'deployment_name',
'value': 'deployment1'
}
]},
}
}
}
# jobの実行
response = requests.post(url+"/v2/jobs/"+job_id+"/runs?project_id="+project_id, headers=headers, json=jobrunpost, verify=False).json()
job_run_id = response['metadata']['asset_id']
job_run_id
# -output-
# 'e0cfc2fd-cf5a-4679-85a2-38855fdc67cb'
実行が成功した場合、Jobの実行IDが出力されます。
##Job実行ステータスを確認
出力されたJobの実行IDからJobの実行ステータスを確認することが可能です。
使用APIはGET /v2/jobs/{job_id}/runs/{run_id}です。
job_run_id = "<Jobの実行ID>"
# Job run status
response = requests.get(url+"/v2/jobs/"+job_id+"/runs/"+job_run_id+"?project_id="+project_id, headers=headers, verify=False).json()
response['entity']['job_run']['state']
# -output-
# 'Completed'
##結果確認
プロジェクトのメイン画面から「ジョブ」タブを押下し、ジョブ一覧から作成したジョブを選択します。
ジョブの詳細画面からスケジュールの実行履歴を確認することができます。
さいごに
こちらの記事作成に際し、以下の記事を参考にさせていただきました。ありがとうございました。
Cloud Pak for Dataの分析プロジェクトのJobを環境変数付きでAPI実行する