一個適合小公司用的 data pipeline 工具

最近在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 裏調整順序就好了

如何運行

摘自官方網站

# 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又能夠無縫鏈接，對咱們的業務有很大的提高。

Prettyyes 不以貌取人最膚淺