气流文件传感器,用于检测本地驱动器上的文件

Question

有没有人对FileSensor有任何想法？我在研究本地目录中的传感文件时遇到了它.代码如下:

  task= FileSensor(
    task_id="senseFile"
    filepath="etc/hosts",
    fs_conn_id='fs_local',
   _hook=self.hook,
    dag=self.dag,)

我还将我的conn_id和conn类型设置为File(path)并给出{'path':'mypath'}但是即使我设置了一个非现有路径,或者如果文件不在指定路径中,则任务完成而dag是成功的.FileSensor似乎根本无法感知文件.

Answer 1

我发现社区贡献了FileSenor有点平庸,所以写了我自己的.

我让它在本地运行服务器/调度程序的文件,但在使用网络路径时遇到了问题.

我找到的网络路径的技巧是将网络驱动器安装到我的Linux Box.

这是我的DAG用于sensor_task >> proccess_task >> archive_task >>触发器重新运行

注意:我们使用通过WebGUI输入的变量(sourcePath,filePattern和archivePath)

    from airflow import DAG
    from airflow.operators import PythonOperator, OmegaFileSensor, ArchiveFileOperator, TriggerDagRunOperator
    from datetime import datetime, timedelta
    from airflow.models import Variable

    default_args = {
        'owner': 'glsam',
        'depends_on_past': False,
        'start_date': datetime(2017, 6, 26),
        'provide_context': True,
        'retries': 100,
        'retry_delay': timedelta(seconds=30)
    }

    task_name = 'my_first_file_sensor_task'
    filepath = Variable.get("soucePath")
    filepattern = Variable.get("filePattern")
    archivepath = Variable.get("archivePath")

    dag = DAG(
        'task_name',
        default_args=default_args,
        schedule_interval=None,
        catchup=False,
        max_active_runs=1,
        concurrency=1)

    sensor_task = OmegaFileSensor(
        task_id=task_name,
        filepath=filepath,
        filepattern=filepattern,
        poke_interval=3,
        dag=dag)


    def process_file(**context):
        file_to_process = context['task_instance'].xcom_pull(
            key='file_name', task_ids=task_name)
        file = open(filepath + file_to_process, 'w')
        file.write('This is a test\n')
        file.write('of processing the file')
        file.close()


    proccess_task = PythonOperator(
        task_id='process_the_file', python_callable=process_file, dag=dag)

    archive_task = ArchiveFileOperator(
        task_id='archive_file',
        filepath=filepath,
        task_name=task_name,
        archivepath=archivepath,
        dag=dag)

    trigger = TriggerDagRunOperator(
        task_id='trigger_dag_rerun', trigger_dag_id=task_name, dag=dag)

    sensor_task >> proccess_task >> archive_task >> trigger

然后这是我的FileSenor

    import os
    import re

    from datetime import datetime
    from airflow.models import BaseOperator
    from airflow.plugins_manager import AirflowPlugin
    from airflow.utils.decorators import apply_defaults
    from airflow.operators.sensors import BaseSensorOperator


    class ArchiveFileOperator(BaseOperator):
        @apply_defaults
        def __init__(self, filepath, archivepath, task_name, *args, **kwargs):
            super(ArchiveFileOperator, self).__init__(*args, **kwargs)
            self.filepath = filepath
            self.archivepath = archivepath
            self.task_name = task_name

        def execute(self, context):
            file_name = context['task_instance'].xcom_pull(self.task_name, key='file_name')
            os.rename(self.filepath + file_name, self.archivepath + file_name)


    class OmegaFileSensor(BaseSensorOperator):
        @apply_defaults
        def __init__(self, filepath, filepattern, *args, **kwargs):
            super(OmegaFileSensor, self).__init__(*args, **kwargs)
            self.filepath = filepath
            self.filepattern = filepattern

        def poke(self, context):
            full_path = self.filepath
            file_pattern = re.compile(self.filepattern)

            directory = os.listdir(full_path)

            for files in directory:
                if not re.match(file_pattern, files):
                    # do nothing
                else:
                    context['task_instance'].xcom_push('file_name', files)
                    return True
            return False

    class OmegaPlugin(AirflowPlugin):
        name = "omega_plugin"
        operators = [OmegaFileSensor, ArchiveFileOperator]

完全免责声明:我是python的完全新手,所以它可能是非常丑陋的代码,但它正在工作