外部から Watson Studio Cloud のジョブを実行

はじめに

Watson Studio CloudのJupyter Notebookジョブ実行。日経平均株価を毎日取得してDb2 on Cloudに保管 に掲載されている通り、Watson Studio Cloud には Jupyter Notebook をそのまま Job として実行できる機能があります。

また、SPSS Modeler Flowでバッチ処理をしよう に記載されている通り、SPSS Modeler Flow で予測モデルを作成し、 Watson Machine Learning のデプロイメント・スペースにプロモートして作成したバッチ処理をそのまま Job として実行できる機能もあります。

一方で、スケジュールに祝日カレンダーを含めたい、成功・警告・失敗時を運用管理用のメールなどに通知したいといった様々な要件を満たすために機能拡張が必要になるケースがあるかと思います。

IBM Workload Scheduler、Apache Airflow、JP1 などの外部の Job scheduler から Watson Studio Cloud のこれらの Job を実行できれば、上記の課題を解決できると思います。

そこで、Watson Data API を使用して Python でサンプルコードを実装してみました。

前提

• 今回は Watson Studio CloudのJupyter Notebookジョブ実行。日経平均株価を毎日取得してDb2 on Cloudに保管 を一通り実行し、Watson Studio Cloud に Job を登録済みであること。
• Python の実行環境が準備済みであること。
  • 今回は Python 3.10 を使用して動作させています。
• Job scheduler の代替として Python の schedule で定期実行します。

作業手順の流れ

1. IBM Cloud の API キーを取得
2. Watson Data API を使用して Watson Studio Cloud のJob を起動
3. Job を schedule に登録して定期実行

作業手順1. IBM Cloud の API キーを取得

1. IBM Cloud にログインします。
2. API キー の画面を表示します。
3. IBM Cloud API キーの作成 ボタンを押下します。
   
4. 名前 と 説明 に任意の文字を反映し、作成 ボタンを押下します。
   
5. ポップアップにて ダウンロード リンクを押下します。
   
   • ⚠️注意 API キーは悪用されると第三者にご自身の IBM Cloud 環境の操作をされてしまう恐れがあるため、厳重に管理してください。
6. ポップアップを閉じます。

作業手順2. Watson Data API を使用して Watson Studio Cloud の Job を起動

1. Python 仮想環境の作成します。

   python3 -m venv venv
   

   source venv/bin/activate
   

2. 必要な python ライブラリーをインストールします。

   pip install requests schedule
   

3. 以下、サンプルコード の関数 job() の変数をご自身の環境に合わせて修正することで Job を実行することができます。
   • api_key: 1. IBM Cloud の API キーを取得 の API キーを設定
   • job_name: 今回は nikkei3 を設定
   • project_id: 以下 URL の projects/ と /jobs の間の文字列を設定
     

   api_key = "<your api key>"
   job_name = "<your job name>"
   project_id = "<your project id>"
   

4. 関数 job() では以下の処理を行なっています。
   • ベアラートークン生成 (関数 generate_bearer_token())
     • 参考: IBM Cloud | API キーを使用した IBM Cloud IAM トークンの生成
   • Job ID 取得 (関数 get_job_id())
     • 参考: Watson Data API | Get list of jobs under a project or a space.
   • Job 実行 (関数 run_job())
     • 参考: Watson Data API | Start a run for a job.
   • Job ステータス確認 (関数 check_job_status())
     • 参考: Watson Data API | Get a specific run of a job.

作業手順3. Job を schedule に登録して定期実行

1. 今回は Job scheduler の代替として Python の schedule で1分毎に Job を実行します。
2. 以下のコードが該当箇所になります。

   schedule.every(1).minutes.do(job)
     while True:
       schedule.run_pending()
       time.sleep(1)
   

3. サンプルコード のスクリプトを実行します。

   python3 path/to/main.py
   

4. 実行結果は以下の通りになります。
   • なお、IBM Cloud の画面操作で該当 Job の実行をキャンセルした場合、Completed の部分が Canceled になります。

   INFO:__main__:Job is working...
   INFO:__main__:Bearer token has been generated.
   INFO:__main__:Job ID is [9f4b81f1-1362-43d4-86bb-78c3418f7364].
   INFO:__main__:Run ID is [53bc2716-dedc-419b-ae6d-e9061fe48aab].
   INFO:__main__:Job status is [Starting].
   INFO:__main__:Job status is [Starting].
   INFO:__main__:Job status is [Starting].
   INFO:__main__:Job status is [Starting].
   INFO:__main__:Job status is [Running].
   INFO:__main__:Job status is [Running].
   INFO:__main__:Job status is [Running].
   INFO:__main__:Job status is [Completed].
   INFO:__main__:Completed.
   
   （切り返し実行）
   

5. 以下の操作で Job 実行履歴の画面を確認をします。
   • ナビゲーション・メニュー を押下
   • プロジェクト > すべてのプロジェクトの表示 リンクを押下
     
   • <ご自身のプロジェクト名> リンクを押下
     
   • ジョブ タブを押下
   • <ご自身のジョブ名> リンクを押下
     
   • ジョブの詳細 にて実行履歴を表示
     

終わりに

今回は Watson Data API にフォーカスしましたが、API & SDK リファレンス・ライブラリー に掲載されている通り、他の API や Java、Node.js、Python、.NET の SDK を使用して開発することができます。なお、現時点で SDK がない API があるのでご留意ください。

参考

• API & SDK リファレンス・ライブラリー
• How to Access Watson Studio Assets by API
• schedule — schedule 1.1.0 documentation

お断り

このサイトの掲載内容は私自身の見解であり、必ずしも所属会社の立場、戦略、意見を代表するものではありません。 記事は執筆時点の情報を元に書いているため、必ずしも最新情報であるとはかぎりません。 記事の内容の正確性には責任を負いません。自己責任で実行してください。