Apache Airflow用のDataRobotプロバイダー¶

DataRobot MLOpsと Apache Airflowの機能を組み合わせることで、モデルの再トレーニングと再デプロイに信頼性の高いソリューションが提供されます。 たとえば、スケジュール、モデルのパフォーマンス低下、または新しいデータがあるときにパイプラインをトリガーするセンサーを使用して、モデルを再トレーニングおよび再デプロイできます。 Apache Airflow用のDataRobotプロバイダーに関するこのクイックスタートガイドでは、基本的な Apache Airflow DAG（有向非巡回グラフ）を実装して、エンドツーエンドのDataRobot AIパイプラインを調整することによるセットアップと設定のプロセスを説明します。 このパイプラインには、プロジェクトの作成、モデルのトレーニング、モデルのデプロイ、予測のスコアリング、ターゲットと特徴量ドリフトデータを返すことなどが含まれます。 さらに、このガイドでは、さまざまなDataRobotパイプラインを迅速に実装できるように、airflow-provider-datarobotリポジトリから サンプルDAGファイルをインポートする方法も説明します。

Apache Airflow用のDataRobotプロバイダーは、 GitHubのパブリックリポジトリで利用可能なソースコードから構築されたPythonパッケージであり、 PyPi（Pythonパッケージインデックス）で公開されています。 これは、 Astronomerレジストリにも一覧表示されています。 プロバイダーパッケージの使用と開発の詳細については、 Apache Airflowのドキュメントを参照してください。 この連携では、REST APIを介してDataRobotインスタンスと通信する DataRobot Python APIクライアントが使用されます。 詳細については、 DataRobot Pythonパッケージのドキュメントを参照してください。

必須コンポーネントのインストール¶

Apache Airflow用のDataRobotプロバイダーには、以下の依存関係がインストールされた環境が必要です。

• Apache Airflow >= 2.3

• Python APIクライアント >= 3.2.0b1

DataRobotプロバイダーをインストールするには、次のコマンドを実行します。

pip install airflow-provider-datarobot 

開始する前に、ローカルのAirflowインスタンスを管理するための、 Astronomerコマンドラインインターフェイス（CLI）ツールをインストールします。

brew install astro 

curl -sSL https://install.astronomer.io | sudo bash 

次に、 pyenvまたは他のPythonバージョンマネージャーをインストールします。

ローカルAirflowプロジェクトの初期化¶

インストールの前提条件が完了したら、新しいディレクトリを作成し、 AstroCLIを使用してローカルのAirflowプロジェクトを初期化できます。

1. 新しいディレクトリを作成し、そのディレクトリに移動します。

   mkdir airflow-provider-datarobot && cd airflow-provider-datarobot 
   

2. 新しいディレクトリ内で次のコマンドを実行し、必要なファイルを含む新しいプロジェクトを初期化します。

   astro dev init 
   

3. requirements.txtファイルに移動し、次のコンテンツを追加します。

   airflow-provider-datarobot 
   

4. 次のコマンドを実行して、DockerコンテナでローカルのAirflowインスタンスを開始します。

   astro dev start 
   

5. インストールが完了し、Webサーバーが起動すると（約1分後）、http://localhost:8080/でAirflowにアクセスできるようになります。

6. Airflowにサインインします。 Airflow DAGページが表示されます。

   

サンプルDAGをAirflowに読み込む¶

DAGの例は、デフォルトでは****DAGページには表示されません。 Apache AirflowのサンプルDAG用のDataRobotプロバイダーを使用できるようにするには：

1. DAGファイルを airflow-provider-datarobotリポジトリからダウンロードします。

2. datarobot_pipeline_dag.pyAirflow DAG（またはdatarobot_provider/example_dagsディレクトリ全体）をプロジェクトにコピーします。

3. 1～2分待ってページを更新します。

   サンプルDAGは、datarobot_pipeline DAGを含むDAGページに表示されます。

   

AirflowからDataRobotへの接続の作成¶

次のステップでは、AirflowからDataRobotへの接続を作成します。

1. 管理 > 接続をクリックして、 Airflow接続を追加します。

2. リスト接続ページで、+ 新しいレコードを追加をクリックします。

3. 接続を追加ダイアログボックスで、次のフィールドを設定します。

   

   フィールド	説明
   接続ID	datarobot_default（すべての演算子でこの名前をデフォルトで使用）
   接続タイプ	DataRobot
   APIキー	DataRobot APIトークン（開発者ツールでAPIキーを配置または作成）
   DataRobotエンドポイントのURL	デフォルトではhttps://app.jp.datarobot.com/api/v2

4. テストをクリックして、AirflowとDataRobotの間にテスト接続を確立します。

5. 接続テストが成功したら、保存をクリックします。

DataRobotパイプラインDAGの設定¶

datarobot_pipeline Airflow DAGには、DataRobotパイプラインのステップを自動化する演算子とセンサーが含まれています。 各演算子は特定のジョブを開始し、各センサーは所定のアクションが完了するまで待機します。

DataRobotパイプラインの各演算子には、特定のパラメーターが必要です。 これらのパラメーターを設定JSONファイルで定義し、DAGの実行時にJSONを指定します。

{
    "training_data": "local-path-to-training-data-or-s3-presigned-url-",
    "project_name": "Project created from Airflow",
    "autopilot_settings": {
        "target": "readmitted",
        "mode": "quick",
        "max_wait": 3600
    },
    "deployment_label": "Deployment created from Airflow",
    "score_settings": {}
} 

autopilot_settingsのパラメーターは Project.set_target()メソッドに直接渡されます。このメソッドで使用可能なパラメーターは、設定JSONファイルを通じて設定することができます。

training_dataおよびscore_settingsの値は、入力/出力タイプによって異なります。 score_settingsのパラメーターは BatchPredictionJob.score()メソッドに直接渡されます。このメソッドで使用可能なパラメーターは、設定JSONファイルを通じて設定することができます。

たとえば、以下のローカルファイル入出力およびAmazon AWS S3入出力JSON設定のサンプルを参照してください。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
    "training_data": "include/Diabetes10k.csv",
    "project_name": "Project created from Airflow",
    "autopilot_settings": {
        "target": "readmitted",
        "mode": "quick",
        "max_wait": 3600
    },
    "deployment_label": "Deployment created from Airflow",
    "score_settings": {}
} 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

{
    "training_data": "include/Diabetes10k.csv",
    "project_name": "Project created from Airflow",
    "autopilot_settings": {
        "target": "readmitted",
        "mode": "quick",
        "max_wait": 3600
    },
    "deployment_label": "Deployment created from Airflow",
    "score_settings": {
        "intake_settings": {
            "type": "localFile",
            "file": "include/Diabetes_scoring_data.csv"
        },
        "output_settings": {
            "type": "localFile",
            "path": "include/Diabetes_predictions.csv"
        }
    }
} 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

{
    "training_data": "s3-presigned-url",
    "project_name": "Project created from Airflow",
    "autopilot_settings": {
        "target": "readmitted",
        "mode": "quick",
        "max_wait": 3600
    },
    "deployment_label": "Deployment created from Airflow",
    "datarobot_aws_credentials": "connection-id",
    "score_settings": {}
} 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

{
    "training_data": "s3-presigned-url",
    "project_name": "Project created from Airflow",
    "autopilot_settings": {
        "target": "readmitted",
        "mode": "quick",
        "max_wait": 3600
    },
    "deployment_label": "Deployment created from Airflow",
    "datarobot_aws_credentials": "connection-id",
    "score_settings": {
        "intake_settings": {
            "type": "s3",
            "url": "s3://path/to/scoring-data/Diabetes10k.csv",
        },
        "output_settings": {
            "type": "s3",
            "url": "s3://path/to/results-dir/Diabetes10k_predictions.csv",
        }
    }
} 

DataRobotパイプラインDAGの実行¶

上記の設定手順を完了すると、組み立てた設定JSONを使用して、AirflowでDataRobotプロバイダーDAGを実行できます。

1. Airflow DAGページで、実行するDAGパイプラインを見つけます。

   

2. そのDAGの実行アイコンをクリックし、設定でDAGをトリガーをクリックします。

   

3. DAG confパラメーターページで、DAGに必要なJSON設定データを入力します。 この例では、前の手順で組み立てたJSONになります。

4. トリガー時にDAGの一時停止を解除を選択し、トリガーをクリックします。 DAGが実行を開始します。