气流 HdfsSensor 操作员不工作答案

【问题标题】：Airflow HdfsSensor operator is not working气流 HdfsSensor 操作员不工作
【发布时间】：2021-02-21 14:03:29
【问题描述】：

我正在尝试使用 Airflow 中的 Hdfs 传感器操作符来触发基于给定路径中的文件到达的下一个任务。但是，当我部署 dag 时，得到的错误为

DAG 损坏：[/usr/local/airflow/dags/test_sensor_dag.py] 语法无效（client.py，第 1473 行） 代码：

from airflow import DAG
from airflow.sensors.hdfs_sensor import HdfsSensor
from datetime import datetime


default_args = {'owner': 'airflow',
                'depends_on_past': False,
                'provide_context': True,
                'start_date': datetime(2020, 3, 19, 0, 0),
                'email': ['hr@***.com'],
                'email_on_failure': True,
                'email_on_retry': False,
                'retries': 0,
                'concurrency': 1
                }

# run it daily at 6AM
schedule_interval = '00 6 * * *'

dag_name = 'test_sensor_dag'

dag = DAG(
    dag_id=dag_name,
    default_args=default_args,
    schedule_interval=schedule_interval)

source_data_sensor = HdfsSensor(
    task_id='source_data_sensor',
    filepath='/data/test/file.csv',
    poke_interval=10,
    timeout=5,
    dag=dag
).poke()

success_notification = EmailOperator(to=['hr@***.com'], task_id='success_notification',
                                     subject='[Success:] test for {{ ds }}',
                                     html_content='Successfully ran the DAG',
                                     dag=dag)

source_data_sensor
success_notification.set_upstream(source_data_sensor)

【问题讨论】：

标签： hdfs airflow-scheduler airflow

【解决方案1】：

您无需在代码中调用.poke()，airflow 会在运行时执行 poke 函数

【讨论】：

【解决方案2】：

在安装气流时运行以下命令序列对我有用：

   pip install snakebite-py3
   pip install azure-storage>=0.34.0,<0.37.0
   pip install 'apache-airflow[all]'

更新：更好的是，如果您不使用 azure，只需从 airflow packages 安装所需的软件包。不过安装snakebite是必须的

【讨论】：

【解决方案3】：

您的问题不是戳或类似的问题，如果您正确安装了 hdfs 提供程序，只需让您的 dag 更简单（只放置您的 hdfs 传感器任务并删除电子邮件内容）并再次运行它，然后提供更多错误。

仅供参考：如您所知，client.py 与 HTTP 和 XMLRPC 相关，因此您只需在 cnf 文件中检查它，并确保您以正确的形式和语法提供了正确的配置，然后再次运行它。还要确保您遵循最佳做法以避免错误的导入。

【讨论】：