はじめに

GKE（Kubernetes）とAirflowの勉強もかねて、GKEでAirflowの環境を構築します。
（ここをかなり参考にしてます。）

Cloud Composerを使わないのか？

※2019/4/16時点
GCPではAirflowのマネージドサービスであるCloud Composerというサービスがあり、
がっつりComposerでジョブ管理していくのであればぜひComposerを使いたい。
が、スモールスタートするには月数万円とそれなりのコストがかかるので
まずはGKEで半マネージドな感じでやってみます。

ちなみに、GKEだとg1-smallで月5千円くらいになります。（GKEは最小3ノード）
f1-microだとさすがにAirflow動かないので、g1-smallが最小インスタンスです。
個人でやってみる場合にはpreemptible（24時間起動後再起動するインスタンス）モードで立ち上げれば月1500円くらいになります。

参考：Composerの料金

Cloud Composer Flexible Environment？

Google Cloud Next '19でComposerについても少し触れられており、
そのうちフレキシブルな環境構築ができるようになるか？

実行環境

諸々環境作るのが面倒なので、Cloud Shell使う前提です。

Airflowの構成

Airflowの構成を確認しつつ、GKE + GCPの各種サービスで以下のような構成を作る。

①Webserver
WebUIを提供するPod。

②Scheduler
ジョブスケジュールに関する制御を行うPod。ちなみにSchedulerのプロセスが動いてないと手動実行もできない（ずっとno statusになる）。

③Worker
実際に処理を実行するプロセス。CeleryExcecuterを使用してWorkerを増やすことで処理の分散ができる。CeleryExcecuter以外のスケールの方法は公式ドキュメント参照

④Database
ジョブの情報やスケジュールの情報を格納。GCEのディスクで永続化。GCEのディスクで永続化すると、ReadWriteOnce となり、1つのノードからしかマウントできない。statefulsetにしておけば、常に同じノードで立ち上がるので、Pod再立ち上げの際にマウントエラーが起きない。
また、各Podからアクセスがくるのでクラスタ内部でアクセスできるようにClusterIPを紐付ける。

⑤Redis
ジョブ実行状態を保持するためのミドルウェア。RabbitMQ等でも良いらしい。ちなみにComposerはRedis使用。
こちらも各Podからアクセスがくるのでクラスタ内部でアクセスできるようにClusterIPを紐付ける。

⑥Service（NodePort）とIngress
外からアクセスできるように、Service（NodePort）とIngressを設定する。
アクセスを制御したい場合は、ロードバランサに対してCloud ArmorでIP制限する等対応する。

⑦GCS（Logs）
ログは各種Podから書き込まれてくるため、GCSに設定。

⑧SouceRepositories（Dags）
AirflowはDagfile（Python）により実行ジョブの情報を管理する。
バージョン管理もできるので、SourceRepositoriesを使用。

参考：Composerの構成

諸々のファイル類

GKEへデプロイするための、諸々のファイル類は以下の通り。
github

airflow-gke
├config
｜｜ airflow.cfg
├keyfile
｜｜サービスアカウントのkeyfile（〜.json）
├script
｜｜ entrypoint.sh
｜ Dockerfile
｜ airflow_deploy.yaml
｜ cloudbuild.yaml
｜ ingress_airflow.yaml

Dockerfile

「set your environment」の部分のENVは使用する自分の環境の情報を記載する

FROM python:3.6

ENV AIRFLOW_GPL_UNIDECODE=yes
ENV CONFIG /root/.config/gcloud
ENV PATH /root/google-cloud-sdk/bin:$PATH
ENV AIRFLOW_HOME /airflow
ENV AIRFLOW_CONFIG $AIRFLOW_HOME/airflow.cfg
ENV AIRFLOW_USER airflow
# ---set your environment
ENV GCPKEY=
ENV GCPSERVICEACCOUNT=
ENV GCPPROJECT=
ENV DAG_REPOSITORY=
ENV AIRFLOW_LOGIN_USER=
ENV AIRFLOW_LOGIN_PASS=
ENV AIRFLOW_FIRSTNAME= 
ENV AIRFLOW_LASTNAME= 
ENV AIRFLOW_EMAIL=
# ---

RUN apt update \
    && apt-get -y install \
                  sudo \
                  lhasa \
                  vim \
                  curl \
                  nmap \
                  netcat\
    && pip install \
                  redis \
                  celery \
                  psycopg2 \
                  apache-airflow \
                  apache-airflow[postgres] \
                  apache-airflow[celery] \
                  apache-airflow[gcp_api] \
                  flask_bcrypt \
                  google-api-python-client \
                  pandas_gbq\
    && useradd -ms /bin/bash -d ${AIRFLOW_HOME} ${AIRFLOW_USER} \
    && curl https://sdk.cloud.google.com | bash    

ADD config/airflow.cfg ${AIRFLOW_HOME}
ADD keyfile/${GCPKEY} ${CONFIG}/${GCPKEY}
ADD script/entrypoint.sh ${AIRFLOW_HOME}

WORKDIR ${AIRFLOW_HOME}

RUN chmod -R 755 ${AIRFLOW_HOME}
RUN chown -R ${AIRFLOW_USER}: ${AIRFLOW_HOME}
RUN chown -R ${AIRFLOW_USER}: /root
RUN echo "%${AIRFLOW_USER} ALL=NOPASSWD: ALL" >> /etc/sudoers

USER ${AIRFLOW_USER}

ENTRYPOINT ["./entrypoint.sh"]

airflow_deploy.yaml

このファイルでDeploymentやらServiceやらまとめて記載している。
Ingressは稼働させておくと料金かかるので、必要なときだけ作成するために別ファイルで管理。

webserverとschedulerとworkerのDockerImageに関しては、コンテナレジストリにPushしたイメージIDを上書きする。

apiVersion : v1
kind : Service
metadata :
  name : service-airflow
spec :
#  type : LoadBalancer
  type : NodePort
  ports :
  - port : 8080 
    targetPort : 8080
  selector : 
    app: airflow-webserver

---
apiVersion: v1
kind: Service
metadata:
  name: postgres
spec:
  type: ClusterIP
  selector:
    app: postgres
  ports:
    - name: postgres
      protocol: TCP
      port: 5432
      targetPort: postgres

---
apiVersion: v1
kind: Service
metadata:
  name: redis
spec:
  type: ClusterIP
  selector:
    app: redis
  ports:
    - name: redis
      protocol: TCP
      port: 6379
      targetPort: redis

---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgres
spec:
  serviceName : postgres
  replicas: 1
  selector :
    matchLabels :
      app : postgres
  template:
    metadata:
      labels:
        app: postgres
    spec:
      containers:
        - name: postgres
          image: postgres
          ports:
            - name: postgres
              containerPort: 5432
          env:
            - name: POSTGRES_USER
              value: "airflow"
            - name: POSTGRES_PASSWORD
              value: "airflow"
            - name: POSTGRES_DB
              value: "airflow"
            - name: PGDATA
              value: "/var/lib/postgresql/data/pgdata"
          volumeMounts :
            - name : volume-postgres
              mountPath : /var/lib/postgresql/data
  volumeClaimTemplates:
    - metadata : 
        name : volume-postgres
      spec : 
        accessModes: [ "ReadWriteOnce" ]
        resources:
          requests:
            storage: 10Gi

---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: redis
spec:
  replicas: 1
  selector :
    matchLabels :
      app : redis
  template:
    metadata:
      labels:
        app: redis
    spec:
      containers:
        - name: redis
          image: redis
          ports:
            - name: redis
              containerPort: 6379

---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-webserver
  labels :
    app : airflow-webserver
spec:
  replicas: 1
  template :
    metadata :
      labels :
        app : airflow-webserver
    spec :
      containers:
        - name: airflow-webserver
          image: #container image
          imagePullPolicy : Always
          ports :
            - containerPort : 8080
          args : ["webserver"]
          readinessProbe:
            httpGet:
              path: /login/
              port: 8080

---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-scheduler
spec:
  replicas: 1
  template :
    metadata :
      labels :
        app : airflow-scheduler
    spec :
      containers:
        - name: airflow-scheduler
          image: #container image
          imagePullPolicy : Always
          args : ["scheduler"]

---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-worker
  labels :
    app : airflow-worker
spec:
  replicas: 1
  template :
    metadata :
      labels :
        app : airflow-worker
    spec :
      containers:
        - name: airflow-worker
          image: #container image
          imagePullPolicy : Always
          args : ["worker"]

個人的ハマりポイント

postgresqlの永続化

postgresqlのstatefulsetのENVでPGDATAを「/var/lib/postgresql/data/pgdata」にした上で
「/var/lib/postgresql/data」をマウントしないとエラーでるか、Pod再作成されたときにリセットされてしまう。
参考

Ingressのヘルスチェック

ヘルスチェックしにいくページのステータスが200で返ってこないとヘルスチェックがエラーになる。
デフォルトは「/」をチェックしにいくが、Airflowでユーザー認証設定してる場合は、リダイレクトが発生しヘルスチェックは「/」のままだとエラーになるので、こちら参考にヘルスチェック先を変更する。
どの時点でステータス200になるかは、こちらでwebserverのserviceを一時的にLoadBalancerタイプにして確認する。

ClusterIPの紐付け

当たり前といえば当たり前だが、各Podとclusterip serviceのlabelを合わせる必要ある
。これ合ってないとservice自体は立ち上がるが、kubectl describe serviceしたときにエンドポイントがnoneになってて結局Pod間通信できない。
色々なWebサイト参考にしながらやるとこの辺が結構不整合起きる。。

Workerの起動

ルートユーザーでWorker起動させようとすると以下メッセージでる。

Running a worker with superuser privileges when the
worker accepts messages serialized with pickle is a very bad idea!

要はルート権限を持つユーザーでWorker起動するなということみたいなので、
airflowというユーザーを作って、そのユーザーで起動する。

ingress_airflow.yaml

apiVersion: extensions/v1beta1
kind: Ingress
metadata:
  name: airflow-ingress
  namespace: default
  annotations:
    kubernetes.io/ingress.global-static-ip-name: "airflow-ingress-ip" # optional
spec:
  backend:
    serviceName: service-airflow 
    servicePort: 8080

cloudbuild.yaml

Cloud Buildを使用する場合には作成する。
個人的にはかなり便利だった。
Source RepositoriesへのpushをトリガーにしてDockerイメージの作成、Container Registryへのpush、Podの再作成など一連の作業を自動化できる。
もちろんトリガーだけでなく、手動でCloud Build実行できる。

steps:
- name: 'gcr.io/cloud-builders/docker'
  args: ['build', '-t', '$_CONTAINER_IMG', '$_FILE_PATH']
- name: 'gcr.io/cloud-builders/gcloud'
  args: ['auth', 'configure-docker']
- name: 'gcr.io/cloud-builders/docker'
  args: ['push', '$_CONTAINER_IMG']
- name: 'gcr.io/cloud-builders/kubectl'
  args: ['delete', '-f', '$_FILE_PATH/$_AIRFLOW_YAML']
  env:
  - 'CLOUDSDK_COMPUTE_ZONE=$_ZONE'
  - 'CLOUDSDK_CONTAINER_CLUSTER=$_CLUSTER_NAME'
- name: 'gcr.io/cloud-builders/kubectl'
  args: ['apply', '-f', '$_FILE_PATH/$_AIRFLOW_YAML']
  env:
  - 'CLOUDSDK_COMPUTE_ZONE=$_ZONE'
  - 'CLOUDSDK_CONTAINER_CLUSTER=$_CLUSTER_NAME'

個人的ハマりポイント

ResponseError: code=403, message=EXTERNAL: Required "container.clusters.get" permission(s) for ~
が出るときは、CloudBuildのサービスアカウントにGKEの管理者の権限を追加する

script

entrypoint.sh

Podが作成されるタイミングで実行されるshellファイル。
やってることは以下。

AirflowのGCPコネクションを作成する
postgresql、redisの起動を待つ
webserverの場合はinitdb（upgradedb）したあとに、airflowのログインユーザー作成する
dagリポジトリのクローン
各サービスの起動

connecitonsを作成する部分の{keyfilename}と{gcp_projectid}はそれぞれ、サービスアカウントのkeyfile名とgcpのプロジェクトIDに置き換える。

# !/usr/bin/env bash

CMD="airflow"
TRY_LOOP="${TRY_LOOP:-10}"
POSTGRES_HOST="${POSTGRES_HOST:-postgres}"
POSTGRES_PORT=5432
REDIS_HOST="${REDIS_HOST:-redis}"
REDIS_PORT=6379

echo "Postgres host: $POSTGRES_HOST"
echo "Redis host: $REDIS_HOST"

# set gcp connections
$CMD connections --add --conn_id=airflow_gcp --conn_type=google_cloud_platform --conn_extra='{"extra__google_cloud_platform__key_path":"/root/.config/gcloud/{keyfilename}","extra__google_cloud_platform__project":"{gcp_projectid}","extra__google_cloud_platform__scope":"https://www.googleapis.com/auth/cloud-platform"}'

# wait for postgres
if [ "$1" = "webserver" ] || [ "$1" = "worker" ] || [ "$1" = "scheduler" ] ; then
  i=0

  while [ `sudo nping --tcp $POSTGRES_HOST -p $POSTGRES_PORT | grep -i Rcvd: | sed s/^.*Rcvd:// | sed s/\(.*// | tr -d ' '` -eq 0 ] 
  do
    i=`expr $i + 1`
    if [ $i -ge $TRY_LOOP ]; then
      echo "$(date) - ${POSTGRES_HOST}:${POSTGRES_PORT} still not reachable, giving up"
      exit 1
    fi
    echo "$(date) - waiting for ${POSTGRES_HOST}:${POSTGRES_PORT}... $i/$TRY_LOOP"
    sleep 5
  done
  
  # initdb and register user
  if [ "$1" = "webserver" ]; then
    echo "Initialize database..."
    $CMD upgradedb

    echo "Register user..."
    $CMD create_user -r Admin -u ${AIRFLOW_LOGIN_USER} -p ${AIRFLOW_LOGIN_PASS} -f ${AIRFLOW_FIRSTNAME} -l ${AIRFLOW_LASTNAME} -e ${AIRFLOW_EMAIL}
  fi
fi

# wait for redis
if [ "$1" = "webserver" ] || [ "$1" = "worker" ] || [ "$1" = "scheduler" ] ; then
  j=0

  while [ `sudo nping --tcp $REDIS_HOST -p $REDIS_PORT | grep -i Rcvd: | sed s/^.*Rcvd:// | sed s/\(.*// | tr -d ' '` -eq 0 ] 
  do
    j=`expr $j + 1`
    if [ $j -ge $TRY_LOOP ]; then
      echo "$(date) - ${REDIS_HOST}:${REDIS_PORT} still not reachable, giving up"
      exit 1
    fi
    echo "$(date) - waiting for ${REDIS_HOST}:${REDIS_PORT}... $j/$TRY_LOOP"
    sleep 5
  done
fi

# git clone dags
gcloud auth activate-service-account ${GCPSERVICEACCOUNT} --key-file $CONFIG/${GCPKEY} --project ${GCPPROJECT}
gcloud source repos clone ${DAG_REPOSITORY} --project=${GCPPROJECT}

# execute args
$CMD "$@"

個人的ハマりポイント

source repositories Request had insufficient authentication scopes.
↑のメッセージが出てくるときは、サービスアカウントにsource repositories への権限を付与する。

config

airflow.cfg

airflowに関するconfigファイル。
かなり長いので重要なところ（デフォルト修正したところ、自分の環境情報を設定するところ）抜粋。

ディレクトリの指定、GCSの指定

[core]
# The home folder for airflow, default is ~/airflow
airflow_home = /airflow #ホームディレクトリ

# The folder where your airflow pipelines live, most likely a
# subfolder in a code repository
# This path must be absolute
dags_folder = /airflow/dags #Dagのディレクトリ

# The folder where airflow should store its log files
# This path must be absolute
base_log_folder = /airflow/logs #logのディレクトリ

# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
# Users must supply an Airflow connection id that provides access to the storage
# location. If remote_logging is set to true, see UPDATING.md for additional
# configuration requirements.
remote_logging = True #ログをGCSに置くのでTrue
remote_log_conn_id = airflow_gcp #GCPへのコネクションID entrypoint.shにて設定
remote_base_log_folder = gs:// #ログを出力するgcsバケット
encrypt_s3_logs = False

executorの指定、DBへの接続情報

# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor
executor = CeleryExecutor #デフォルトはSequentialExecutor（WebUIから特定ジョブのリランできない）

# The SqlAlchemy connection string to the metadata database.
# SqlAlchemy supports many different database engine, more information
# their website
sql_alchemy_conn = postgresql+psycopg2://airflow:airflow@postgres:5432/airflow #posgresへの接続情報 デフォルトはSQLiteになってる（CeleryExecutor使えない）

Airflowのユーザー認証

# Set to true to turn on authentication:
# https://airflow.incubator.apache.org/security.html#web-authentication
authenticate = True  #デフォルトはFalse
auth_backend = airflow.contrib.auth.backends.password_auth

Admin権限のユーザーを作成するための設定

# Use FAB-based webserver with RBAC feature
rbac = True #デフォルトはFalse

エラー時のメール送信元設定

[smtp]
# If you want airflow to send emails on retries, failure, and you want to use
# the airflow.utils.email.send_email_smtp function, you have to configure an
# smtp server here
smtp_host = smtp.gmail.com #gmailのsmtpサーバ
smtp_starttls = True
smtp_ssl = False
smtp_user =  #ログインユーザ
smtp_password = #ログインパス（アプリパスワードなので注意！）
# Uncomment and set the user/pass settings if you want to use SMTP AUTH
# smtp_user = airflow
# smtp_password = airflow
smtp_port = 587 #gmailのsmtpサーバのポート
smtp_mail_from =  #送信元メールアドレス

Celeryのbroker（Redis）と実行結果先（postgresql）の接続情報

# The Celery broker URL. Celery supports RabbitMQ, Redis and experimentally
# a sqlalchemy database. Refer to the Celery documentation for more
# information.
# http://docs.celeryproject.org/en/latest/userguide/configuration.html#broker-settings
broker_url = redis://redis:6379/1

# The Celery result_backend. When a job finishes, it needs to update the
# metadata of the job. Therefore it will post a message on a message bus,
# or insert it into a database (depending of the backend)
# This status is used by the scheduler to update the state of the task
# The use of a database is highly recommended
# http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-result-backend-settings
result_backend = db+postgresql://airflow:airflow@postgres:5432/airflow

keyfile

GCPのサービスアカウントのkeyファイル（〜.json）を配置する用のディレクトリ。

GKEへデプロイ

GKEクラスタ立ち上げ

「airflow-gke」というクラスタを立ち上げる
以下は、g1-smallの10GBディスクでus-central1-aゾーンに3ノード構成でクラスタを作成するサンプルコマンド。

サンプル

gcloud container clusters create airflow-gke --machine-type=g1-small --num-nodes=3 --disk-size=10 --zone=us-central1-a

preemptibleにする場合は、「--preemptible」オプションを追加すればよい

kubectlの認証情報の設定

kubernetesクラスタは基本的に「kubectl」コマンドで操作するが、
どのクラスタに対してのコマンド発行なのかを認証する

gcloud container clusters get-credentials airflow-gke --zone=us-central1-a

リモートリポジトリのclone

任意のディレクトリで以下実行

git clone https://github.com/yakamazu/airflow-gke.git

Source RepositoriesでDag用のリポジトリを作成

詳細割愛
任意のDag用のリポジトリを作成する。

ログ用のGCSバケットを作成

詳細割愛
任意のログ用バケットを作成する。

サービスアカウントの作成、keyfileの配置

詳細割愛
keyfileはgcsにアップしてgsutil使ってcloudshell環境に持っていく等。

Dockerfileの修正

---set your environmentの部分
ENV GCPKEY：サービスアカウントのキーファイル名
ENV GCPSERVICEACCOUNT：サービスアカウント名
ENV GCPPROJECT：GCPプロジェクトID
ENV DAG_REPOSITORY：SourceRepositoriesのDAGのリポジトリ名
ENV AIRFLOW_LOGIN_USER：AirflowのWebUIログインユーザー
ENV AIRFLOW_LOGIN_PASS：AirflowのWebUIログインパスワード
ENV AIRFLOW_FIRSTNAME：AirflowのWebUIログインユーザー表示名（名）
ENV AIRFLOW_LASTNAME：AirflowのWebUIログインユーザー表示名（姓）
ENV AIRFLOW_EMAIL：AirflowのWebUIログインユーザーのメールアドレス

configファイルの修正

諸々のファイル類の部分で記載したconfigを修正する。
特にログのGCSの場所と、メール情報。

entrypoint.shの修正

connecitonsを作成する部分の{keyfilename}と{gcp_projectid}はそれぞれ、サービスアカウントのkeyfile名とgcpのプロジェクトIDに置き換える。

$CMD connections --add --conn_id=airflow_gcp --conn_type=google_cloud_platform --conn_extra='{"extra__google_cloud_platform__key_path":"/root/.config/gcloud/{keyfilename}","extra__google_cloud_platform__project":"{gcp_projectid}","extra__google_cloud_platform__scope":"https://www.googleapis.com/auth/cloud-platform"}'

Dockerイメージの作成

Dockerfileが格納されているディレクトリで以下コマンド実行。
airflow-gkeというイメージが作成される。

docker image build -t asia.gcr.io/プロジェクトID/airflow-gke:latest .

Container RegistryへのPush

docker push 使うための認証

gcloud auth configure-docker

Container RegistryへのPush

docker push asia.gcr.io/プロジェクトID/airflow-gke:latest

airflow_deploy.yamlの修正

deploymentのwebserverとschedulerとworkerに関しては、コンテナレジストリにPushしたイメージIDを上書きする。
「image」の#container imageのところを「asia.gcr.io/プロジェクトID/airflow-gke:latest」にする。

airflow_deploy.yamlのapply

以下コマンドの実行

kubectl apply -f airflow_deploy.yaml

Ingressに固定IPを設定する

こちら参考にingress_airflow.yamlのglobal-static-ip-nameと名前を合わせた固定IPを作成する。

gcloud compute addresses create airflow-ingress-ip --global

ingress_airflow.yamlのapply

以下コマンドの実行
※起動するとロードバランサー立ち上がり、料金かかる

kubectl apply -f ingress_airflow.yaml

起動確認

Podの一覧

1/1のRunningになってればOK

$ kubectl get pod 

NAME                                 READY     STATUS    RESTARTS   AGE
airflow-scheduler-7bc759f895-vvssx   1/1       Running   0          13m
airflow-webserver-6bd76d6d6b-lwzhw   1/1       Running   0          13m
airflow-worker-59d954659d-v2bb4      1/1       Running   0          13m
postgres-0                           1/1       Running   0          12m
redis-56c6759df8-clfqj               1/1       Running   0          13m

Serviceの一覧

postgresとredisとservice-airflowが立ち上がってればOK

$ kubectl get service

NAME              TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)          AGE
kubernetes        ClusterIP   10.3.240.1    <none>        443/TCP          9h
postgres          ClusterIP   10.3.254.25   <none>        5432/TCP         14m
redis             ClusterIP   10.3.249.32   <none>        6379/TCP         14m
service-airflow   NodePort    10.3.243.40   <none>        8080:30957/TCP   14m

Ingressの一覧

airflow-ingressが立ち上がっていて、作成した外部IP（airflow-ingress-ip）が割り当てられていればOK
IPが割り当てられてからアクセス可能になるまでは若干タイムラグあり。

$ kubectl get ingress
NAME              HOSTS     ADDRESS          PORTS     AGE
airflow-ingress   *         ~.~.~.~          80        5h

DAGの登録、反映

DAGをSourceRepositoriesにPushし、Airflowに更新を反映させるには、
サービスを再起動させる必要がある。
各Podでサービス再起動してもよいが面倒なので、ingress_airflow.yamlをdeleteしてからapplyする。

kubectl delete -f airflow_deploy.yaml
kubectl apply -f airflow_deploy.yaml

WebUIログイン

IngressのIPにアクセスし、Dockerfileで設定したAirflowログインユーザー名、パスワードを入力する。

※sampleのDagファイルは自分で予め作ってSource RepositoriesにPushしておく。

ジョブ実行

sampleのDAGを実行。

エラー時メール配信

エラーになったらこんな感じでメール届く。

ジョブリラン

GCSにログ出力

指定したバケットにDAG単位でフォルダできる

DAGフォルダの中はタスクID単位でのフォルダができる

タスクIDフォルダの中に実行時間ごとのフォルダができる

実行時間フォルダの中にログ出力される

（おまけ）よく使うkubectlコマンド

kubectl get pod -o wide

podの一覧を表示する。
-o wideをつけると、IPやどのNodeで立ち上がってるかの情報も出力される。

kubectl exec -it pod名 /bin/bash

podの中に入るイメージ。

kubectl get service

serviceの一覧を表示する。IPを確認できる。

kubectl get ingress

ingressの一覧を表示する。IPを確認できる。

kubectl get pvc

pvc（永続ボリューム）の一覧を表示する。

kubectl describe pod pod名

podの詳細情報、起動時のメッセージを確認できる。

kubectl describe service service名

serviceの詳細情報を確認できる。

kubectl logs pod名

podの標準出力を確認する。

kubectl apply -f yamlファイル

yamlファイルの内容をデプロイする

kubectl delete -f yamlファイル

yamlファイルの内容を削除する