airflow简单了解

python 评论

Q：airflow介绍

是用于任务调度和监控的开源工具，由python编写。以DAG有向无环图的方式组建task任务流，灵活的定义任务间的依赖关系，监控任务的运行状态。强大的web UI，页面功能丰富。

任务流怎么理解，工作流程有一个明确的开始和结束，按一定的间隔运行，就可以用python定义一个dag文件。
dag的定义文件，只是一个配置文件，该脚本不能用于任务之间的交叉通信，不过有一个高级特性解决该问题XComs。
下面是定义任务依赖关系的方法：

  • t1.set_upstream(t2)
  • t1.set_downstream([t2, t3])
  • t1 >> [t2, t3]

官方的一个简单的dag文件定义如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from datetime import datetime, timedelta
from textwrap import dedent

# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG

# Operators; we need this to operate!
from airflow.operators.bash import BashOperator
with DAG(
    "tutorial",
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "email": ["airflow@example.com"],
        "email_on_failure": False,
        "email_on_retry": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'sla': timedelta(hours=2),
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function,
        # 'on_success_callback': some_other_function,
        # 'on_retry_callback': another_function,
        # 'sla_miss_callback': yet_another_function,
        # 'trigger_rule': 'all_success'
    },
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        depends_on_past=False,
        bash_command="sleep 5",
        retries=3,
    )
    t1.doc_md = dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](http://montcs.bloomu.edu/~bobmon/Semesters/2012-01/491/import%20soul.png)
    **Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    templated_command = dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id="templated",
        depends_on_past=False,
        bash_command=templated_command,
    )

    t1 >> [t2, t3]

Q：airflow基本功能

  1. 任务调度：支持按时间计划、依赖关系或手动触发等方式调度任务。

  2. 任务监控：可以查看任务执行情况，如执行时间、任务状态、失败原因等。

  3. 依赖关系管理：支持设置任务间的依赖关系，确保任务的执行顺序。

  4. 数据流图：提供数据流图，方便用户清晰地查看任务间的依赖关系。

  5. 任务执行：支持将任务提交到不同的执行环境，如本地、云环境等。

  6. 高度定制：支持自定义插件，以扩展Airflow的功能。

  7. 权限管理：支持对用户进行权限管理，确保任务的安全性。

  8. Web界面：提供易于使用的Web界面，方便用户进行任务调度、监控、配置等操作。

  9. 任务扩展：支持任务扩展，可以根据需求增加任务类型。

  10. 数据库存储：支持使用数据库存储任务信息，方便用户对任务信息进行查询和维护。

  11. 任务记录：支持对任务的执行情况进行详细记录，以便用户查询和分析。

  12. 电子邮件通知：支持对任务的执行情况进行电子邮件通知，以便用户及时了解任务执行情况。

  13. 多租户支持：支持多租户，可以将不同的任务隔离开来，保证各个任务的安全性和独立性。

  14. 可靠性：支持多种高可靠性策略，如任务重试、任务隔离等，确保任务的可靠性。

  15. 安全性：支持多种安全措施，如用户认证、数据加密等，确保系统的安全性。

  16. 故障恢复：支持故障恢复，在系统出现故障时，可以自动恢复任务的执行。

  17. API支持：提供丰富的API，方便用户通过代码编程方式对任务进行操作。

  18. 可扩展性：提供可扩展的架构，可以根据业务需求进行扩展。

  19. 报表生成：支持报表生成，方便用户查询和分析任务执行情况。

  20. 日志管理：支持日志管理，方便用户查询和分析系统运行情况。

  21. 可视化管理：提供可视化的任务管理界面，方便用户对任务进行监控和管理。

  22. 支持多种语言：支持多种语言，如Python、Java、C++等，方便用户编写任务代码。

    Q：airflow应用场景

  23. 数据管道：用于编排数据管道，从数据源获取数据，并在进行数据处理、转换、存储等操作后将数据存储到目标数据库。

  24. 数据分析：用于编排数据分析任务，将数据从数据库中抽取，并进行数据分析、可视化等操作。

  25. 模型训练：用于编排模型训练任务，将数据从数据库中抽取，并使用机器学习模型进行训练。

  26. 定时任务：用于编排定时任务，定期执行一些固定的任务，如数据备份、定期清理等。

  27. 其他自动化任务：用于编排其他自动化任务，如文件转换、数据同步等。

    Q：airflow有哪些组件

    image.png

  • Webserver：管理界面，可以在其中管理任务、查看任务执行状态、配置任务等。
  • Scheduler：调度器，负责调度任务执行。
  • Workers：执行任务的节点，负责执行任务。
  • DAGs：任务流程，表示任务之间的依赖关系。
  • Operators：任务执行单元，封装了一组任务执行的逻辑。
  • Hooks：连接器，用于连接不同的任务或外部系统。
  • Sensors：传感器，用于监测外部系统的状态。
  • Variables：变量，存储一些全局变量，可以在任务流程中使用。
  • XComs：交互数据，可以在任务之间传递数据。
  • Connections：连接，用于连接不同的数据源。

    Q：airflow和dolphinsheduler的不同和优缺点

    DolphinScheduler致力于可视化，通过页面完成一个DAG工作流。
    Airflow则是通过编程的方式完成一个DAG工作量，定制化开发相对比较容易。

相似点：

  • 两者都允许用户创建和管理复杂的工作流。
  • 两者都有基于网页的用户界面，用于查看和管理作业。
  • 两者都提供了一种自动调度和执行任务的方法。

不同点：

  • Airflow 是用 Python 编写的，具有庞大的活跃社区，而 DolphinScheduler 是用 Java 编写的，社区较小。
  • Airflow 有丰富的内置操作符，用于处理数据，例如在系统之间传输文件的能力，并且可以使用自定义插件轻松扩展，而 DolphinScheduler 提供了一组更基本的操作符，可能需要更多的自定义才能适合特定的用例。
  • Airflow 的监控和报警功能更强大，相比之下，DolphinScheduler 提供了基本的监控和报警功能。
  • Airflow 具有更灵活的执行模型，允许以顺序或并行的方式执行任务，而 DolphinScheduler 主要支持顺序执行。

Airflow 的优点：

  • 功能丰富：Airflow 提供了诸如 DAG 编排，任务审核，任务监控等功能，并且具有很好的扩展性，可以使用第三方插件扩展其功能。
  • 用户友好：Airflow 提供了一个简单易用的 Web 界面，方便用户进行任务管理和监控。
  • 集成性强：Airflow 在整个 DevOps 生命周期中有良好的集成性，可以方便地与各种数据存储，数据分析，机器学习等工具集成。

Airflow 的缺点：

  • 部署复杂：Airflow 的部署相对比较复杂，对于新手可能有一定的学习曲线。
  • 资源消耗大：由于 Airflow 提供了大量的功能，因此对于资源的消耗也相对较大。

DolphinScheduler 的优点：

  • 部署简单：DolphinScheduler 的部署相对比较简单，易于上手。
  • 资源消耗小：DolphinScheduler 对于资源的消耗相对较小，适合在资源有限的情况下使用。
  • 稳定性高：DolphinScheduler 具有较高的稳定性，可以保证任务的顺利执行。

DolphinScheduler 的缺点：

  • 功能较弱：相对于 Airflow，DolphinScheduler 的功能相对较弱，可能无法满足复杂的任务管理需求。
  • 可扩展性差：DolphinScheduler 可扩展性较差，可能不支持用户希望的一些特殊功能。

HAHADeveloper & Designer

个人简介