一個(gè)適合小公司用的 data pipeline 工具

2i18ns 發(fā)布于2019-07-24 18:19 / 2643人閱讀

摘要：有了自己的系統(tǒng)我覺(jué)得就很安心了，以后能夠做數(shù)據(jù)處理和機(jī)器學(xué)習(xí)方面就相對(duì)方便一些。隆重推薦的工具是我很喜歡的公司，他們有很多開(kāi)源的工具，我覺(jué)得是最實(shí)用的代表。是，在很多機(jī)器學(xué)習(xí)里有應(yīng)用，也就是所謂的有向非循環(huán)。

最近在Prettyyes一直想建立起非常專(zhuān)業(yè)的data pipeline系統(tǒng)，然后沒(méi)有很多時(shí)間，這幾個(gè)禮拜正好app上線(xiàn)，有時(shí)間開(kāi)始建立自己的 data pipeline，能夠很好的做每天的數(shù)據(jù)導(dǎo)入，數(shù)據(jù)收集，以及數(shù)據(jù)分析。

什么是ETL

ETL 是常用的數(shù)據(jù)處理，在以前的公司里，ETL 差不多是數(shù)據(jù)處理的基礎(chǔ)，要求非常穩(wěn)定，容錯(cuò)率高，而且能夠很好的監(jiān)控。ETL的全稱(chēng)是 Extract，Transform，Load，一般情況下是將亂七八糟的數(shù)據(jù)進(jìn)行預(yù)處理，然后放到儲(chǔ)存空間上。可以是SQL的也可以是NOSQL的，還可以直接存成file的模式。

一開(kāi)始我的設(shè)計(jì)思路是，用幾個(gè)cron job和celery來(lái)handle所有的處理，然后將我們的log文件存在hdfs，還有一些數(shù)據(jù)存在mysql，大概每天跑一次。核心是能夠scale，穩(wěn)定，容錯(cuò)，roll back。我們的data warehouse就放在云上，就簡(jiǎn)單處理了。

有了自己的ETL系統(tǒng)我覺(jué)得就很安心了，以后能夠做數(shù)據(jù)處理和機(jī)器學(xué)習(xí)方面就相對(duì)方便一些。

問(wèn)題來(lái)了

一開(kāi)始我設(shè)計(jì)的思路和Uber一開(kāi)始的ETL很像，因?yàn)槲矣X(jué)得很方便。但是我發(fā)覺(jué)一個(gè)很?chē)?yán)重的問(wèn)題，我一個(gè)人忙不過(guò)來(lái)。首先，要至少寫(xiě)個(gè)前端UI來(lái)監(jiān)控cron job，但是市面上的都很差。其次，容錯(cuò)的autorestart寫(xiě)起來(lái)很費(fèi)勁，可能是我自己沒(méi)有找到一個(gè)好的處理方法。最后部署的時(shí)候相當(dāng)麻煩，如果要寫(xiě)好這些東西，我一個(gè)人的話(huà)要至少一個(gè)月的時(shí)間，可能還不是特別robust。在嘗試寫(xiě)了2兩天的一些碎片處理的腳本之后我發(fā)覺(jué)時(shí)間拖了實(shí)在太久了。

隆重推薦的工具

airbnb是我很喜歡的公司，他們有很多開(kāi)源的工具，airflow我覺(jué)得是最實(shí)用的代表。airflow 是能進(jìn)行數(shù)據(jù)pipeline的管理，甚至是可以當(dāng)做更高級(jí)的cron job 來(lái)使用。現(xiàn)在一般的大廠(chǎng)都說(shuō)自己的數(shù)據(jù)處理是ETL，美其名曰 data pipeline，可能跟google倡導(dǎo)的有關(guān)。airbnb的airflow是用python寫(xiě)的，它能進(jìn)行工作流的調(diào)度，提供更可靠的流程，而且它還有自帶的UI（可能是跟airbnb設(shè)計(jì)主導(dǎo)有關(guān)）。話(huà)不多說(shuō)，先放兩張截圖：

什么是DAG

airflow里最重要的一個(gè)概念是DAG。

DAG是directed asyclic graph，在很多機(jī)器學(xué)習(xí)里有應(yīng)用，也就是所謂的有向非循環(huán)。但是在airflow里你可以看做是一個(gè)小的工程，小的流程，因?yàn)槊總€(gè)小的工程里可以有很多“有向”的task，最終達(dá)到某種目的。在官網(wǎng)中的介紹里說(shuō)dag的特點(diǎn)：

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超級(jí)簡(jiǎn)單，使用pip就可以，現(xiàn)在airflow的版本是1.6.1，但是有個(gè)小的bug，這個(gè)之后會(huì)告訴大家如何修改。

pip install airflow

這里有個(gè)坑，因?yàn)閍irflow涉及到很到數(shù)據(jù)處理的包，所以會(huì)安裝pandas和numpy（這個(gè)Data Scientist應(yīng)該都很熟悉）但是國(guó)內(nèi)pip install 安裝非常慢，用douban的源也有一些小的問(wèn)題。我的解決方案是，直接先用豆瓣的源安裝numpy 和 pandas，然后再安裝airflow，自動(dòng)化部署的時(shí)候可以在requirements.txt 里調(diào)整順序就行了

如何運(yùn)行

摘自官方網(wǎng)站

# 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，來(lái)監(jiān)控你的進(jìn)程。

如何導(dǎo)入dag

一般第一次運(yùn)行之后，airflow會(huì)在默認(rèn)文件夾下生成airflow文件夾，然后你只要在里面新建一個(gè)文件dag就可以了。我這邊部署在阿里云上的文件tree大概是這個(gè)樣子的。

以下是我自己寫(xiě)的我們公司prettyyes里需要每天處理log的其中一個(gè)小的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
)

寫(xiě)好之后，只要將這個(gè)dag放入之前建立好的dag文件夾，然后運(yùn)行：

python

來(lái)確保沒(méi)有語(yǔ)法錯(cuò)誤。在測(cè)試?yán)锬憧梢钥吹轿业?/p>

schedule_interval=timedelta(days=1)

這樣我們的數(shù)據(jù)處理的任務(wù)就相當(dāng)于每天跑一次。更重要的是，airflow還提供處理bash處理的接口外還有hadoop的很多接口。可以為以后連接hadoop系統(tǒng)提供便利。很多具體的功能可以看官方文檔。

其中的一個(gè)小的bug

airflow 1.6.1有一個(gè)網(wǎng)站的小的bug，安裝成功后，點(diǎn)擊dag里的log會(huì)出現(xiàn)以下頁(yè)面：

這個(gè)只要將

airflow/www/utils.py

文件替換成最新的airflow github上的utils.py文件就行，具體的問(wèn)題在這個(gè)：

fixes datetime issue when persisting logs

使用supervisord進(jìn)行deamon

airflow本身沒(méi)有deamon模式，所以直接用supervisord就ok了，我們只要寫(xiě)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

我覺(jué)得airflow特別適合小的團(tuán)隊(duì)，他的功能強(qiáng)大，而且真的部署方便。和hadoop，mrjob又可以無(wú)縫連接，對(duì)我們的業(yè)務(wù)有很大的提升。

Prettyyes 不以貌取人最膚淺