apache-spark - 气流 SparkKubernetesOperator 日志记录

Question

我在 Airflow 中使用 KubernetesExecutor 作为 Executor。我的 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 执行成功。我可以通过执行在 spark-driver 日志中看到输出kubectl logs spark-pi-driver

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

Answer 1

更新 SparkKubernetesSensor 配置如下

t2 = SparkKubernetesSensor(task_id='spark_pi_monitor',

命名空间=“默认”，

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

达格=达格，

attach_log=真，)