在當(dāng)今的大數(shù)據(jù)世界中，應(yīng)用程序產(chǎn)生大量的電子數(shù)據(jù) – 這些巨大的電子數(shù)據(jù)存儲(chǔ)庫包含了有價(jià)值的、寶貴的信息。 對于人類分析師或領(lǐng)域?qū)＜?，很難做出有趣的發(fā)現(xiàn)或?qū)ふ铱梢詭椭鷽Q策過程的模式。 我們需要自動(dòng)化的流程來有效地利用龐大的，信息豐富的數(shù)據(jù)進(jìn)行規(guī)劃和投資決策。 在處理數(shù)據(jù)之前，收集數(shù)據(jù)，聚合和轉(zhuǎn)換數(shù)據(jù)是絕對必要的，并最終將數(shù)據(jù)移動(dòng)到那些使用不同分析和數(shù)據(jù)挖掘工具的存儲(chǔ)庫中。

執(zhí)行所有這些步驟的流行工具之一是Apache Flume。 這些數(shù)據(jù)通常是以事件或日志的形式存儲(chǔ)。 Apache Flume有三個(gè)主要組件：

• Source：數(shù)據(jù)源可以是企業(yè)服務(wù)器，文件系統(tǒng)，云端，數(shù)據(jù)存儲(chǔ)庫等。
• Sink：Sink是可以存儲(chǔ)數(shù)據(jù)的目標(biāo)存儲(chǔ)庫。 它可以是一個(gè)集中的地方，如HDFS，像Apache Spark這樣的處理引擎，或像ElasticSearch這樣的數(shù)據(jù)存儲(chǔ)庫/搜索引擎。
• Channel：在事件被sink消耗前由Channel 存儲(chǔ)。 Channel 是被動(dòng)存儲(chǔ)。 Channel 支持故障恢復(fù)和高可靠性; Channel 示例是由本地文件系統(tǒng)和基于內(nèi)存的Channel 支持的文件通道。

Flume是高度可配置的，并且支持許多源，channel，serializer和sink。它還支持?jǐn)?shù)據(jù)流。 Flume的強(qiáng)大功能是攔截器，支持在運(yùn)行中修改/刪除事件的功能。支持的攔截器之一是regex_filter。

regex_filter將事件體解釋為文本，并將其與提供的正則表達(dá)式進(jìn)行對比，并基于匹配的模式和表達(dá)式，包括或排除事件。我們將詳細(xì)看看regex_filter。

要求

從數(shù)據(jù)源中，我們以街道號(hào)，名稱，城市和角色的形式獲取數(shù)據(jù)。現(xiàn)在，數(shù)據(jù)源可能是實(shí)時(shí)流數(shù)據(jù)，也可能是任何其他來源。在本示例中，我已經(jīng)使用Netcat服務(wù)作為偵聽給定端口的源，并將每行文本轉(zhuǎn)換為事件。要求以文本格式將數(shù)據(jù)保存到HDFS中。在將數(shù)據(jù)保存到HDFS之前，必須根據(jù)角色對數(shù)據(jù)進(jìn)行過濾。只有經(jīng)理的記錄需要存儲(chǔ)在HDFS中;其他角色的數(shù)據(jù)必須被忽略。例如，允許以下數(shù)據(jù)：

1,alok,mumbai,manager 
 
2,jatin,chennai,manager  

下列的數(shù)據(jù)是不被允許的：

3,yogesh,kolkata,developer 
 
5,jyotsana,pune,developer  

如何達(dá)到這個(gè)要求

可以通過使用 regex_filter 攔截器來實(shí)現(xiàn)。這個(gè)攔截器將根據(jù)規(guī)則基礎(chǔ)來進(jìn)行事件過濾，只有感興趣的事件才會(huì)發(fā)送到對應(yīng)的槽中，同時(shí)忽略其他的事件。

## Describe regex_filter interceptor and configure exclude events attribute 
 
a1.sources.r1.interceptors = i1 
 
a1.sources.r1.interceptors.i1.type = regex_filter 
 
a1.sources.r1.interceptors.i1.regex = developer 
 
a1.sources.r1.interceptors.i1.excludeEvents = true  

HDFS 槽允許數(shù)據(jù)存儲(chǔ)在 HDFS 中，使用文本/序列格式。也可以使用壓縮格式存儲(chǔ)。

a1.channels = c1 
 
a1.sinks = k1 
 
a1.sinks.k1.type = hdfs 
 
a1.sinks.k1.channel = c1 
 
## assumption is that Hadoop is CDH 
 
a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 
 
a1.sinks.k1.hdfs.fileType= DataStream 
 
a1.sinks.k1.hdfs.writeFormat = Text  

如何運(yùn)行示例

首先，你需要 Hadoop 來讓示例作為 HDFS 的槽來運(yùn)行。如果你沒有一個(gè) Hadoop 集群，可以將槽改為日志，然后只需要啟動(dòng) Flume。 在某個(gè)目錄下存儲(chǔ) regex_filter_flume_conf.conf 文件然后使用如下命令運(yùn)行代理。

flume-ng agent --conf conf --conf-file regex_filter_flume_conf.conf --name a1 -Dflume.root.logger=INFO,console 

注意代理名稱是 a1。我用了 Netcat 這個(gè)源。

a1.sources.r1.type = netcat 
 
a1.sources.r1.bind = localhost 
 
a1.sources.r1.port = 44444  

一旦 Flume 代理啟動(dòng)，運(yùn)行下面命令用來發(fā)送事件給 Flume。

telnet localhost 40000 

現(xiàn)在我們只需要提供如下輸入文本：

1,alok,mumbai,manager 
 
2,jatin,chennai,manager 
 
3,yogesh,kolkata,developer 
 
4,ragini,delhi,manager 
 
5,jyotsana,pune,developer 
 
6,valmiki,banglore,manager  

訪問 HDFS 你會(huì)觀察到 HDFS 在 hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 下創(chuàng)建了一個(gè)文件，文件只包含經(jīng)理的數(shù)據(jù)。

完整的 flume 配置 — regex_filter_flume_conf.conf — 如下：

# Name the components on this agent 
 
a1.sources = r1 
 
a1.sinks = k1 
 
a1.channels = c1 
 
# Describe/configure the source - netcat 
 
a1.sources.r1.type = netcat 
 
a1.sources.r1.bind = localhost 
 
a1.sources.r1.port = 44444 
 
# Describe the HDFS sink 
 
a1.channels = c1 
 
a1.sinks = k1 
 
a1.sinks.k1.type = hdfs 
 
a1.sinks.k1.channel = c1 
 
a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 
 
a1.sinks.k1.hdfs.fileType= DataStream 
 
a1.sinks.k1.hdfs.writeFormat = Text 
 
## Describe regex_filter interceptor and configure exclude events attribute 
 
a1.sources.r1.interceptors = i1 
 
a1.sources.r1.interceptors.i1.type = regex_filter 
 
a1.sources.r1.interceptors.i1.regex = developer 
 
a1.sources.r1.interceptors.i1.excludeEvents = true 
 
# Use a channel which buffers events in memory 
 
a1.channels.c1.type = memory 
 
a1.channels.c1.capacity = 1000 
 
a1.channels.c1.transactionCapacity = 100 
 
# Bind the source and sink to the channel 
 
a1.sources.r1.channels = c1 
 
a1.sinks.k1.channel = c1  

完整的項(xiàng)目代碼請看 這里。