Airflow

Question

我已经使用 KubernetesPodOperator 在 GCP 运行中上传了包含我需要的所有内容的图像，我收到以下消息，有人可以帮助我了解发生了什么吗？

以下是我的脚本和错误消息的摘要：

import os
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from airflow.contrib.operators.mssql_to_gcs import MsSqlToGoogleCloudStorageOperator
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
import pyarrow
import airflow
from airflow import DAG
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator


default_args = {
    'owner': 'me',
    'start_date': airflow.utils.dates.days_ago(0),
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'depends_on_past': False,
    'catchup': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG('test_kube', default_args=default_args, description='Kubernetes Operator',
         schedule_interval='00 12 01 * *') as dag:
    k = KubernetesPodOperator(namespace='kubenm',
                              image="teste-kube:latest",
                              name="test",
                              task_id="test",
                              is_delete_operator_pod=False,
                              hostnetwork=False,
                              dag=dag
                              )

    k

第一次使用这个算子，不知道能不能满足我的需求

日志：

INFO - Job 11344: Subtask test Traceback (most recent call last):
INFO - Job 11344: Subtask test   File "/usr/local/bin/airflow", line 32, in <module>
INFO - Job 11344: Subtask test     args.func(args)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper
INFO - Job 11344: Subtask test     return f(*args, **kwargs)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/bin/cli.py", line 522, in run
INFO - Job 11344: Subtask test     _run(args, dag, ti)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/bin/cli.py", line 440, in _run
INFO - Job 11344: Subtask test     pool=args.pool,
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/utils/db.py", line 74, in wrapper
INFO - Job 11344: Subtask test     return func(*args, **kwargs)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 926, in _run_raw_task
INFO - Job 11344: Subtask test     result = task_copy.execute(context=context)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/contrib/operators/kubernetes_pod_operator.py", line 111, in execute
INFO - Job 11344: Subtask test     config_file=self.config_file)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/contrib/kubernetes/kube_client.py", line 56, in get_kube_client
INFO - Job 11344: Subtask test     return _load_kube_config(in_cluster, cluster_context, config_file)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/airflow/contrib/kubernetes/kube_client.py", line 38, in _load_kube_config
INFO - Job 11344: Subtask test     config.load_kube_config(config_file=config_file, context=cluster_context)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 645, in load_kube_config
INFO - Job 11344: Subtask test     persist_config=persist_config)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 613, in _get_kube_config_loader_for_yaml_file
INFO - Job 11344: Subtask test     **kwargs)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 153, in __init__
INFO - Job 11344: Subtask test     self.set_active_context(active_context)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 173, in set_active_context
INFO - Job 11344: Subtask test     context_name = self._config['current-context']
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 495, in __getitem__
INFO - Job 11344: Subtask test     v = self.safe_get(key)
INFO - Job 11344: Subtask test   File "/usr/local/lib/python3.7/site-packages/kubernetes/config/kube_config.py", line 491, in safe_get
INFO - Job 11344: Subtask test     key in self.value):
INFO - Job 11344: Subtask test TypeError: argument of type 'NoneType' is not iterable
INFO - [[34m2019-09-30 17:18:16,274[0m] {[34mlocal_task_job.py:[0m172} WARNING[0m - State of this instance has been externally set to [1mup_for_retry[0m. Taking the poison pill.[0m
INFO - Sending Signals.SIGTERM to GPID 9
INFO - Process psutil.Process(pid=9, status='terminated') (9) terminated with exit code -15
INFO - [[34m2019-09-30 17:18:16,303[0m] {[34mlocal_task_job.py:[0m105} INFO[0m

Answer 1

我对脚本进行了一些调整，使操作生效：

with DAG('test_kube', default_args=default_args, description='Kubernetes Operator',
         schedule_interval='00 12 01 * *') as dag:
    k = KubernetesPodOperator(namespace='kubenm',
                              image="gcr.io/project/teste-kube:latest", #Image path was incorrect 
                              name="test",
                              in_cluster=True, #To trigger cluster kubeconfig.
                              image_pull_policy="Always",  #In my case, I need the image update to occur whenever there is an update
                              task_id="test",
                              is_delete_operator_pod=False,
                              hostnetwork=False,
                              dag=dag
                              )

    k

Airflow - KubernetesPodOperator - TypeError: argument of type 'NoneType' is not iterable

Airflow - KubernetesPodOperator - TypeError: argument of type 'NoneType' is not iterable

kubernetes

kubernetes-pod