Python 依存関係のインストール

このページでは、いくつかの一般的なアプリケーションから Python パッケージをインストールして Cloud Composer 環境に接続する方法について説明します。

オプション

依存関係は、基本環境に含まれている既存の Python 依存関係とともにインストールされます。

Python の依存関係に外部の依存関係が含まれておらず、Cloud Composer の依存関係と競合しない場合は、GCP Console、Cloud SDK、Cloud Composer API を使用して Python Package Index から Python 依存関係をインストールできます。

その他の要件がある場合は、次のいくつかのオプションから選択できます。

オプション 使用条件
ローカルの Python ライブラリ Python の依存関係を Python Package Index で見つけることができず、そのライブラリに dist-packages のような外部の依存関係が含まれていない場合。
プラグイン機能 Airflow ウェブ インターフェースの変更など、プラグイン固有の機能を使用する場合。
PythonVirtualenvOperator Python の依存関係が Python Package Index にあり、その依存関係に外部の依存関係が含まれていない場合。しかし、Python の依存関係をすべてのワーカーにインストールする必要がない、または依存関係が Cloud Composer に必要な依存関係と競合する。
KubernetesPodOperator

pip からインストールできない外部の依存関係(dist-packages など)、または内部の pip サーバー上にある外部の依存関係が必要な場合。

このオプションには多くの設定とメンテナンスが必要なため、一般的に、その他のオプションがうまくいかない場合にこれを考慮します。

始める前に

  • Cloud Composer 環境に Python パッケージをインストールするには、composer.environments.update 権限が必要になります。詳細については、Cloud Composer のアクセス制御をご覧ください。
  • 要件は PEP-508 で指定された形式に従う必要があります。各要件は、小文字で指定され、オプションの追加情報とバージョン指定子とともにパッケージ名で構成されます。
  • API を使用してカスタム Python 依存関係をインストールすると、新しくインストールされた PyPI 依存関係を使用して Cloud Composer のすべてのプロセスが実行されます。
  • カスタムの PyPI 依存関係を使用すると、Airflow に必要な依存関係との競合が生じて不安定になる場合があります。
  • 本番環境にデプロイする前に、Airflow ワーカー コンテナ内で PyPI パッケージをローカルにテストすることをおすすめします。

PyPi から Python 依存関係をインストールする

環境の Python 依存関係を追加、更新、削除するには、次の手順を行います。

Console

次のようにパッケージ名とバージョン指定子を指定します。

  • "pi-python-client", "==1.1.post1"
  • "go-api-python-client", "==1.0.0.dev187"

バージョン指定子のないパッケージの場合は、"glob2", " " などの値に空の文字列を使用します。

次の手順で PyPi 依存関係ページに移動して、環境の Python 依存関係にアクセスします。

  1. Google Cloud Platform Console で [環境] ページを開きます。

    [環境] ページを開く

  2. Python 依存関係をインストール、更新、削除する環境の [名前] をクリックします。

  3. [PyPi 依存関係] タブを選択します。

  4. [編集] ボタンをクリックします。

  5. 新しい依存関係を追加するには、次の手順を行います。

    1. [依存関係の追加] ボタンをクリックします。

    2. [名前] フィールドと [バージョン] フィールドに、それぞれライブラリの名前とバージョンを入力します。

  6. 既存の依存関係を更新するには、次の手順を行います。

    1. 更新するライブラリの [名前] フィールド、または [バージョン] フィールドを選択します。

    2. 新しい値を入力します。

  7. 依存関係を削除するには、次の手順を行います。

    1. 削除する依存関係の名前にカーソルを合わせます。

    2. 表示されたゴミ箱アイコンをクリックします。

gcloud

requirements.txt ファイルを gcloud コマンドライン ツールに渡します。それぞれ必要な指定子を使用してファイルを別の行にフォーマットします。

requirements.txt ファイルの例:

scipy>=0.13.3
scikit-learn
nltk[machine_learning]

requirements.txt ファイルを environments.set-python-dependencies コマンドに渡して、インストールの依存関係を設定します。

gcloud composer environments update ENVIRONMENT-NAME \\
--update-pypi-packages-from-file requirements.txt \\
--location LOCATION

オペレーションが完了すると、コマンドは終了します。待機しないようにするには、--async フラグを指定してください。

依存関係の競合が原因で更新が失敗した場合は、既存の依存関係内で環境が引き続き実行します。オペレーションが成功すると、DAG に新しくインストールされた Python 依存関係の使用を開始できます。

REST

updateMask クエリ パラメータの接頭辞として config.softwareConfig.pypiPackages を指定して、projects.locations.environments.patch メソッドを使用します。

ローカルの Python ライブラリをインストールする

社内またはローカルの Python ライブラリをインストールするには、次の手順を行います。

  1. 依存関係を dags/ フォルダのサブディレクトリに配置します。サブディレクトリからモジュールをインポートするには、モジュールのパス内の各サブディレクトリに __init__.py パッケージのマーカー ファイルが含まれている必要があります。

    この例で、依存関係は coin_module.py です。

    dags/
      use_local_deps.py  # A DAG file.
      dependencies/
        __init__.py
        coin_module.py
    
  2. DAG 定義ファイルから依存関係をインポートします。

    例:

    from dependencies import coin_module

Flower ウェブ インターフェースに接続する

Flower は、Celery クラスタを操作するためのウェブベースのツールです。Flower は環境にプリインストールされています。ウェブ UI を使用して、環境の Apache Airflow ワーカーをモニタリングできます。

Flower にアクセスするには、次の手順を行います。

  1. 環境を表示して、Kubernetes Engine クラスタを確認します。

    gcloud composer environments describe ENVIRONMENT-NAME /
        --location LOCATION

    クラスタは gkeCluster としてリストされます。クラスタがデプロイされているゾーンが location として表示されます。

    例:

          gcloud composer environments describe environment-name --location us-central1
          config:
            airflowUri: https://uNNNNe0aNNbcd3fff-tp.appspot.com
            dagGcsPrefix: gs://us-central1-may18-test-00a47695-bucket/dags
            gkeCluster: projects/example-project/zones/us-central1-a/clusters/us-central1-environment-name-00a47695-gke
            nodeConfig:
              diskSizeGb: 100
              location: projects/example-project/zones/us-central1-a

    この例では、クラスタが us-central1-environment-name-00a47695-gke、ゾーンが us-central1-a です。この情報は GCP Console の [環境の詳細] ページでも確認できます。

  2. Kubernetes Engine クラスタに接続します。

    gcloud container clusters get-credentials CLUSTER_NAME /
    --zone CLUSTER_ZONE

    例:

    gcloud container clusters get-credentials us-central1-environment-name-00a47695-gke --zone us-central1-a
    Fetching cluster endpoint and auth data.
    kubeconfig entry generated for us-central1-environment-name-00a47695-gke.
  3. ワーカーポッドを表示して、Flower を実行するポッドを選択します。

    kubectl get pods

    例:

    kubectl get pods
    NAME                                 READY     STATUS    RESTARTS   AGE
    airflow-redis-67f555bdb8-n6m9k       1/1       Running   0          13d
    airflow-scheduler-6cdf4f4ff7-dm4dm   2/2       Running   0          1h
    airflow-sqlproxy-54497bd557-nlqtg    1/1       Running   0          13d
    airflow-worker-c5c4b58c7-bl5bf       2/2       Running   0          1h
    airflow-worker-c5c4b58c7-szqhm       2/2       Running   0          1h
    airflow-worker-c5c4b58c7-zhmkv       2/2       Running   0          1h

    ポッド名は、正規表現 "airflow-(worker|scheduler)-[-a-f0-9]+") と一致します。

  4. このワーカーポッドで Flower を実行します。

    kubectl exec -it POD_NAME -c airflow-worker -- celery flower /
        --broker=redis://airflow-redis-service:6379/0 --port=5555

    例:

    kubectl exec -it airflow-worker-c5c4b58c7-zhmkv -c airflow-worker -- celery flower
    --broker=redis://airflow-redis-service:6379/0 --port=5555
    [I 180601 20:35:55 command:139] Visit me at http://localhost:5555
    [I 180601 20:35:55 command:144] Broker: redis://airflow-redis-service:6379/0
  5. 別のターミナル セッションで、ローカルポートを Flower に転送します。

    kubectl port-forward POD_NAME 5555

    例:

    kubectl port-forward airflow-worker-c5c4b58c7-zhmkv 5555
    Forwarding from 127.0.0.1:5555 -> 5555
  6. ローカル ブラウザで http://localhost:5555 にアクセスして、ウェブ UI にアクセスします。

SQLAlchemy をインストールして Airflow データベースにアクセスする

SQLAlchemy は、Python SQL ツールキットと Object Relational Mapper(ORM)で構成されます。SQLAlchemy をインストールすると、Cloud Composer の Cloud SQL インスタンスにアクセスできます。インストール中に、Cloud Composer により Airflow 環境変数 AIRFLOW__CORE__SQL_ALCHEMY_CONN が構成されます。

SQL Alchemy をインストールするには、次の手順を行います。

  1. 環境に sqlalchemy をインストールします。

    gcloud composer environments update ENVIRONMENT-NAME /
        --location LOCATION /
        --update-pypi-package "sqlalchemy"
    
  2. 環境を表示して、Kubernetes Engine クラスタを確認します。

    gcloud composer environments describe ENVIRONMENT-NAME /
        --location LOCATION
  3. Kubernetes Engine クラスタに接続します。

    gcloud container clusters get-credentials CLUSTER_NAME /
        --zone CLUSTER_LOCATION
  4. ワーカーポッドを表示して、接続するポッドを選択します。

    kubectl get pods
  5. SSH でワーカーポッドに接続します。

    kubectl exec -it POD_NAME `-- /bin/bash`

    例:

    kubectl exec -it airflow-worker-54c6b57789-66pnr -- /bin/bash
    Defaulting container name to airflow-worker.
    Use 'kubectl describe pod/airflow-worker-54c6b57789-66pnr' to see all of the containers in this pod.
    airflow@airflow-worker-54c6b57789-66pnr:/$

  6. sqlalchemy ライブラリを使用して、Airflow データベースとやり取りします。

    python
    import airflow.configuration as config
    config.conf.get('core', 'sql_alchemy_conn')
このページは役立ちましたか?評価をお願いいたします。

フィードバックを送信...