一個適合小公司用的開源數據管道工具
最近在Prettyyes一直想建立起非常專業的data pipeline系統，然后沒有很多時間，這幾個禮拜正好app上線，有時間開始建立自己的 data pipeline，能夠很好的做每天的數據導入，數據收集，以及數據分析。
什么是ETL
ETL 是常用的數據處理，在以前的公司里，ETL 差不多是數據處理的基礎，要求非常穩定，容錯率高，而且能夠很好的監控。ETL的全稱是 Extract，Transform，Load， 一般情況下是將亂七八糟的數據進行預處理，然后放到儲存空間上。可以是SQL的也可以是NOSQL的，還可以直接存成file的模式。
一開始我的設計思路是，用幾個cron job和celery來handle所有的處理，然后將我們的log文件存在hdfs，還有一些數據存在mysql，大概每天跑一次。核心是能夠scale，穩定，容錯，roll back。我們的data warehouse就放在云上，就簡單處理了。
有了自己的ETL系統我覺得就很安心了，以后能夠做數據處理和機器學習方面就相對方便一些。
問題來了
一開始我設計的思路和Uber一開始的ETL很像，因為我覺得很方便。但是我發覺一個很嚴重的問題，我一個人忙不過來。首先，要至少寫個前端UI來監控cron job，但是市面上的都很差。其次，容錯的autorestart寫起來很費勁，可能是我自己沒有找到一個好的處理方法。***部署的時候相當麻煩，如果要寫好這些東西，我一個人的話要至少一個月的時間，可能還不是特別robust。在嘗試寫了2兩天的一些碎片處理的腳本之后我發覺時間拖了實在太久了。
隆重推薦的工具
airbnb是我很喜歡的公司，他們有很多開源的工具，airflow我覺得是最實用的代表。airflow 是能進行數據pipeline的管理，甚至是可以當做更高級的cron job 來使用。現在一般的大廠都說自己的數據處理是ETL，美其名曰 data pipeline，可能跟google倡導的有關。airbnb的airflow是用python寫的，它能進行工作流的調度，提供更可靠的流程，而且它還有自帶的UI（可能是跟airbnb設計主導有關）。話不多說，先放兩張截圖：
什么是DAG
airflow里最重要的一個概念是DAG。
DAG是directed asyclic graph，在很多機器學習里有應用，也就是所謂的有向非循環。但是在airflow里你可以看做是一個小的工程，小的流程，因為每個小的工程里可以有很多“有向”的task，最終達到某種目的。在官網中的介紹里說dag的特點：
- Scheduled: each job should run at a certain scheduled interval
- Mission critical: if some of the jobs aren’t running, we are in trouble
- Evolving: as the company and the data team matures, so does the data processing
- Heterogenous: the stack for modern analytics is changing quickly, and most companies run multiple systems that need to be glued together
- YEAH! It's awesome, right? After reading all of these, I found it's perfectly fit Prettyyes.
如何安裝
安裝airflow超級簡單，使用pip就可以，現在airflow的版本是1.6.1，但是有個小的bug，這個之后會告訴大家如何修改。pip install airflow
這里有個坑，因為airflow涉及到很到數據處理的包，所以會安裝pandas和numpy（這個Data Scientist應該都很熟悉）但是國內pip install 安裝非常慢，用douban的源也有一些小的問題。我的解決方案是，直接先用豆瓣的源安裝numpy 和 pandas，然后再安裝airflow，自動化部署的時候可以在requirements.txt 里調整順序就行了
如何運行
- pip install airflow
摘自官方網站
- # airflow needs a home, ~/airflow is the default,
- # but you can lay foundation somewhere else if you prefer
- # (optional)
- export AIRFLOW_HOME=~/airflow
- # install from pypi using pip
- pip install airflow
- # initialize the database
- airflow initdb
- # start the web server, default port is 8080
- airflow webserver -p 8080
然后你就可以上web ui查看所有的dags，來監控你的進程。
如何導入dag
一般***次運行之后，airflow會在默認文件夾下生成airflow文件夾，然后你只要在里面新建一個文件dag就可以了。我這邊部署在阿里云上的文件tree大概是這個樣子的。
以下是我自己寫的我們公司prettyyes里需要每天處理log的其中一個小的dag：
- from airflow import DAG
- from airflow.operators import BashOperator
- from datetime import datetime, timedelta
- import ConfigParser
- config = ConfigParser.ConfigParser()
- config.read('/etc/conf.ini')
- WORK_DIR = config.get('dir_conf', 'work_dir')
- OUTPUT_DIR = config.get('dir_conf', 'log_output')
- PYTHON_ENV = config.get('dir_conf', 'python_env')
- default_args = {
- 'owner': 'airflow',
- 'depends_on_past': False,
- 'start_date': datetime.today() - timedelta(days=1),
- 'retries': 2,
- 'retry_delay': timedelta(minutes=15),
- }
- dag = DAG('daily_process', default_args=default_args, schedule_interval=timedelta(days=1))
- templated_command = "echo 'single' | {python_env}/python {work_dir}/mr/LogMR.py"\
- .format(python_env=PYTHON_ENV, work_dir=WORK_DIR) + " --start_date {{ ds }}"
- task = BashOperator(
- task_id='process_log',
- bash_command=templated_command,
- dag=dag
- )
寫好之后，只要將這個dag放入之前建立好的dag文件夾，然后運行：
- python <dag_file>
來確保沒有語法錯誤。在測試里你可以看到我的
- schedule_interval=timedelta(days=1)
這樣我們的數據處理的任務就相當于每天跑一次。更重要的是，airflow還提供處理bash處理的接口外還有hadoop的很多接口。可以為以后連接hadoop系統提供便利。很多具體的功能可以看官方文檔。
其中的一個小的bug
airflow 1.6.1有一個網站的小的bug，安裝成功后，點擊dag里的log會出現以下頁面：
這個只要將
- airflow/www/utils.py
文件替換成***的airflow github上的utils.py文件就行，具體的問題在這個：
fixes datetime issue when persisting logs
使用supervisord進行deamon
airflow本身沒有deamon模式，所以直接用supervisord就ok了，我們只要寫4行代碼。
- [program:airflow_web]
- command=/home/kimi/env/athena/bin/airflow webserver -p 8080
- [program:airflow_scheduler]
- command=/home/kimi/env/athena/bin/airflow scheduler
我覺得airflow特別適合小的團隊，他的功能強大，而且真的部署方便。和hadoop，mrjob又可以無縫連接，對我們的業務有很大的提升。
