气流 SparkKubernetesOperator 日志记录

Question

我在 Airflow 中使用 KubernetesExecutor 作为执行器。 我的 DAG 代码

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator
from airflow.providers.cncf.kubernetes.sensors.spark_kubernetes import SparkKubernetesSensor


dag = DAG(
   'spark_pi_using_spark_operator',
   default_args={'max_active_runs': 1},
   description='submit spark-pi as sparkApplication on kubernetes',
   schedule_interval=timedelta(days=1),
   start_date=datetime(2021, 1, 1),
   catchup=False,
)

t1 = SparkKubernetesOperator(
   task_id='spark_pi_submit',
   namespace="default",
   application_file="example_spark_kubernetes_spark_pi.yaml",
   do_xcom_push=True,
   dag=dag,
)

t2 = SparkKubernetesSensor(
   task_id='spark_pi_monitor',
   namespace="default",
   application_name="{{ task_instance.xcom_pull(task_ids='spark_pi_submit')['metadata']['name'] }}",
   dag=dag,
)
t1 >> t2 

DAG 执行成功。我可以通过执行 kubectl logs spark-pi-driver 在 spark-driver 日志中看到输出

但是我无法在 Airflow UI 中看到相同的日志。

Answer 1

如下更新 SparkKubernetesSensor 配置

t2 = SparkKubernetesSensor(
   task_id='spark_pi_monitor',

   namespace="default",

   application_name="{task_instance.xcom_pull(task_ids='spark_pi_submit') 
                      ['metadata']['name'] }}",

   dag=dag,

   attach_log=True,
)