AirFlow調度平臺簡介

airflow 是一個編排、調度和監控工作流的平臺，由Airbnb開源，現在在Apache Software Foundation 孵化。airflow將工作流編排為tasks組成的有向無環圖(DAGs)，調度器在一組workers上按照指定的依賴關系執行tasks。同時，airflow提供了豐富的命令行工具和簡單易用的用戶界面以便用戶查看和操作，并且airflow提供了監控和報警系統

AirFlow基礎概念

Airflow主要是將工作流的相關信息定義到一個Python文件中，airflow根據文件中的定義信息執行工作流，在Airflow pipeline定義中，主要涉及兩個類： DAG，Operator

DAG : 有向無環圖，它將定義的任務按照依賴關系組織起來

Operator：用來描述每個任務具體做的事，airflow內置了很多operator，如BashOperator 執行一個bash 命令，PythonOperator 調用任意的Python 函數，EmailOperator 用于發送郵件，HTTPOperator 用于發送HTTP請求， SqlOperator 用于執行SQL命令…同時，用戶可以自定義Operator，這給用戶提供了極大的便利性。

通過DAG和Operator結合起來就可以構建復雜的工作流了

Talend簡介

Talend是一個開源的ELT任務構建工具，可以通過簡單拖拽的方式設計復雜的ETL任務并自動生成Java代碼，設計完成后可以通過構建導出ETL任務java源碼和可直接執行jar包及執行jar包的shell和bat文件。

接下來，我們通過Talend 桌面軟件構建一個簡單的ELT任務，然后通過airflow UI執行這個ETL任務

Airflow調度ETL任務實例

1.通過Talend 設計、構建ETL任務

打開Talend設計如下簡單的ETL任務，將一個CSV文件導入到數據庫中：

然后，將這個ETL任務構建并導出，目錄結構如下：

我們使用在linux環境中執行shell文件的方式執行ETL任務。

2.定義Airflow pipeline

我們在airflow 指定的dags目錄下新建python文件來定義airflow pipeline

代碼如下:

import airflow

from datetime import datetime

from airflow.operators.bash_operator import BashOperator

from airflow.models import DAG

from airflow.utils.email import send_email

#shell命令執行.sh文件

job_commd = "/home/airflow/jobs/AirFlowTestJob_0.1/AirFlowTestJob/AirFlowTestJob_run.sh '/home/airflow/jobs/AirFlowTestJob_0.1/AirFlowTestJob/'"

#dag中的一些參數設置

default_args = {

'owner': 'Airflow',

'depends_on_past': False,

'start_date': datetime(2018, 7, 23),

'email': ['VALID_EMAIL_ID'],

'email_on_failure': True,

'email_on_success': True,

'retries':3,

}

dag = DAG('csv_to_data', default_args=default_args)

#定義ETL任務,多個任務時定義多個Operator

t1 = BashOperator(

task_id='csv_to_database',

bash_command=job_commd,

dag=dag

)

3.調度執行ETL任務

完成以上工作，我們啟動airflow webserver服務器后,便可以在airflow UI界面看到我們定義的ETl任務csv_to_data，可以進行相應操作。

總結

以上是生活随笔為你收集整理的airflow实现Java定时任务,AirFlow定时调度执行Talend ETL任务的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。