機器學習中的數據預處理有哪些常見/重要的工具？

01-02

想要為以後可能要做的事情做下準備，到目前為止做過分析拿到手的還都是經過一串pipeline預處理過的比較乾淨的數據，想要知道搭建這些pipeline一般需要哪些工具。
比如，未經處理的機器生成的文本log一類，要處理成可以進行分析、建模的形式一般需要哪些步驟，哪些工具是比較必要的？

最近在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 was 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 &

來確保沒有語法錯誤。在測試里你可以看到我的

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](https://github.com/airbnb/airflow/commit/c4d0fe93ed8185114364984e5eafa1b1fc404a0c)

使用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 不以貌取人最膚淺](https://itunes.apple.com/us/app/prettyyes-hao-da/id1053374824?ls=1mt=8)

不太清楚你說的是搭建pipeline的工具還是說pipeline裡面處理數據的工具，就順道都說一下。

1. pipeline工具本身一般是控制這些工具的流程，最簡單的crontab就定時執行就好，但是有時候會有數據依賴的問題，比如第7步依賴第三步的兩個文件以及平行的第6步的文件，這個依賴並不是線性的，而是一個圖的形式。當中加上有些技術延遲比如有些數據生成失敗了需要重啟某些特定的步驟重新生成，這個問題就稍微複雜了。crontab一般在這時候就搞不定，需要一些專門的pipeline管理，比如 spotify/luigi · GitHub 或者 Azkaban

2. 預處理文本數據本身一般用一些Linux的工具比如cut啊awk啊等等做數據篩選和清理，自己寫一寫python小工具做數據的簡單組合比如複雜的regex規則組合，比較大的數據歸類和抽象一般用Hive之類的hadoop工具都可以，裡面也可以插入linux小工具和自己寫的工具。

工業界的數據項目多數時間要設計如何清理數據，這和學術界的玩具數據玩起來很不一樣，歡迎來到真實的世界。

用熟用好python， matlab。

在做數據預處理之前明確目標和應用場景是什麼，至於具體的細節還是別偷懶老老實實編代碼吧。

我理解的數據預處理的主要任務是處理缺失值，光滑雜訊數據，識別或刪除利群點，解決數據的不一致性。

通常比較容易的實現方式是用MATLAB, R 和Python等自己寫，不過對於不熟悉代碼的人來說可能要花點功夫去學習一下。

你可以了解一下這個產品，Elemental Data Engine，目前是公測階段，可以免費註冊使用。簡單的拖拖拽拽就可以實現從數據處理到建模的流程，就算不會寫代碼也可以實現機器學習。

數據預處理需要根據數據本身的特性進行，不同的格式，不同的要求，有缺失值的要填，有無效數據的要剔，有冗餘維的要選，這些步驟都跟數據本身的特性緊密相關，並沒有什麼現成的黑箱工具可以用（當然在一些特定領域針對某種類型的數據會有現成的）。我個人而言做預處理一般就用MATLAB, R 和Python，每次拿到數據，根據數據特徵，矩陣涉及的多不多啊數據量大不大跑起來費不費時啊，需要用到工具包的多不多啊等等，在三個語言里選一個，自己寫腳本做預處理。

如果是你說的預處理是指數據的清洗、數字化、normalization的話，我用的Python，這些步驟大部分可以自己寫、不難；用工具的話，pandas不錯，靈活、全面，包括平均數、方差、補全等都有；在scikit learn庫中也有一些API專門做這個的。

目前在機器學習上只用過Python，其他語言不知道。另外，有些演算法不在意是否數字化和normalization。有時候用hash把數字和字元都映射成數字也可以，簡單省事。

我目前只知道這麼多，希望有幫助。

簡單說一下，目前的演算法都有各種應用，只是行業不同而已哦。

按照機器學習的分法，最常見的分類就是有監督學習、無監督學習、半監督學習、強化學習。

有監督學習主要有：邏輯回歸(Logistic Regression)、BP神經網路(Back Propagation Neural Network)

無監督學習主要有：Apriori、k-means.

半監督學習主要是分類和回歸，有：推理演算法（Graph Inference）、拉普拉斯支持向量機（Laplacian SVM）

強化學習有：Q-learning、時間差學習（Tempral difference learning）

如果按照數據挖掘來分：

分類與回歸：決策樹、貝葉斯、人工神經網路、K-近鄰、支持向量機、邏輯回歸等

關聯：Apriori、FP-樹頻集等

聚類：k-means、k-medoids等

簡單就這麼看看吧，功能這一塊不好說，不同演算法的應用真的不一樣，回頭我整理一下寫到博客裡面。

自然語言處理（比如分詞，去掉連接詞等）的話，有Python的nltk, 收費的Stanford Parser, Java的OpenNLP等等。

數據處理的話，工具更多，商用的如STS, 需要編程的如Weka, R語言等都可以。

谷歌一下很容易搜出幾十個工具來，挨個試試research一下，找出最適合自己的。既然幹了數據挖掘就要心細不怕苦。

圖像的話，先濾個波。

把編程練熟一點，處理任何格式的數據不都隨意