Hadoop辅助工具——Flume、Sqoop
前言
在一個完整的離線大數據處理系統中，除了hdfs+mapreduce+hive組成分析系統的核心之外，還需要數據采集、結果數據導出、任務調度等不可或缺的輔助系統，而這些輔助工具在hadoop生態體系中都有便捷的開源框架，如圖所示：
1. Flume日志采集框架
1.1 Flume介紹
1.1.1 概述
- Flume是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸的系統。
- Flume可以采集文件，socket數據包、文件、文件夾、kafka等各種形式源數據，又可以將采集到的數據(下沉sink)輸出到HDFS、hbase、hive、kafka等眾多外部存儲系統中
- 一般的采集需求，通過對flume的簡單配置即可實現
- Flume針對特殊場景也具備良好的自定義擴展能力
因此，flume可以適用于大部分的日常數據采集場景。
1.1.2 運行機制
a)?Source：采集組件，用于跟數據源對接，以獲取數據
b)?Sink：下沉組件，用于往下一級agent傳遞數據或者往最終存儲系統傳遞數據
c)?Channel：傳輸通道組件，用于從source將數據傳遞到sink
?
1.1.3 Flume采集系統結構圖
?1. 簡單結構——單個agent采集數據
?
2. 復雜結構——多級agent之間串聯
?
?
1.2 Flume實戰案例
1.2.1 Flume的安裝部署
1、Flume的安裝非常簡單，只需要解壓即可，當然，前提是已有hadoop環境。上傳安裝包到數據源所在節點上
然后解壓 ?tar -zxvf apache-flume-1.6.0-bin.tar.gz
然后進入flume的目錄，修改conf下的flume-env.sh，在里面配置JAVA_HOME
2、根據數據采集的需求配置采集方案，描述在配置文件中(文件名可任意自定義)
3、指定采集方案配置文件，在相應的節點上啟動flume agent
先用一個最簡單的例子來測試一下程序環境是否正常
1、先在flume的conf目錄下新建一個配置文件（采集方案）
vi ??netcat-logger.properties
# 定義這個agent中各組件的名字 a1.sources = r1 a1.sinks = k1 a1.channels = c1# 描述和配置source組件：r1 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444# 描述和配置sink組件：k1 a1.sinks.k1.type = logger# 描述和配置channel組件，此處使用是內存緩存的方式 a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100# 描述和配置source channel sink之間的連接關系 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c12、啟動agent去采集數據
bin/flume-ng agent -c conf -f conf/netcat-logger.conf -n a1 -Dflume.root.logger=INFO,console-c conf ??指定flume自身的配置文件所在目錄
-f conf/netcat-logger.con ?指定我們所描述的采集方案
-n a1 ?指定我們這個agent的名字
3、測試
先要往agent的source所監聽的端口上發送數據，讓agent有數據可采。隨便在一個能跟agent節點聯網的機器上。
telnet anget-hostname ?port ??（telnet localhost 44444）?
1.2.2 采集案例
1、采集日志目錄中的文件到HDFS
結構示意圖：
采集需求：某服務器的某特定目錄下，會不斷產生新的文件，每當有新文件出現，就需要把文件采集到HDFS中去
根據需求，首先定義以下3大要素
spooldir特性：
1、監視一個目錄，只要目錄中出現新文件，就會采集文件中的內容
2、采集完成的文件，會被agent自動添加一個后綴：COMPLETED
3、所監視的目錄中不允許重復出現相同文件名的文件
配置文件編寫：
#定義三大組件的名稱 agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1# 配置source組件 agent1.sources.source1.type = spooldir agent1.sources.source1.spoolDir = /home/hadoop/logs/ agent1.sources.source1.fileHeader = false#配置攔截器 agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = host agent1.sources.source1.interceptors.i1.hostHeader = hostname# 配置sink組件 agent1.sinks.sink1.type = hdfs agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = access_log agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text agent1.sinks.sink1.hdfs.rollSize = 102400 agent1.sinks.sink1.hdfs.rollCount = 1000000 agent1.sinks.sink1.hdfs.rollInterval = 60 #agent1.sinks.sink1.hdfs.round = true #agent1.sinks.sink1.hdfs.roundValue = 10 #agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # Use a channel which buffers events in memory agent1.channels.channel1.type = memory agent1.channels.channel1.keep-alive = 120 agent1.channels.channel1.capacity = 500000 agent1.channels.channel1.transactionCapacity = 600# Bind the source and sink to the channel agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1Channel參數解釋：
capacity：默認該通道中最大的可以存儲的event數量
trasactionCapacity：每次最大可以從source中拿到或者送到sink中的event數量
keep-alive：event添加到通道中或者移出的允許時間
?
測試階段，啟動flume agent的命令：
bin/flume-ng agent -c ./conf -f ./dir-hdfs.conf -n agent1 -Dflume.root.logger=DEBUG,console-D后面跟的是log4j的參數，用于測試觀察
生產中，啟動flume，應該把flume啟動在后臺：
nohup bin/flume-ng agent -c ./conf -f ./dir-hdfs.conf -n agent1 1>/dev/null 2>&1 &2、采集文件到HDFS
采集需求：比如業務系統使用log4j生成的日志，日志內容不斷增加，需要把追加到日志文件中的數據實時采集到hdfs
根據需求，首先定義以下3大要素
配置文件編寫：
agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1# Describe/configure tail -F source1 agent1.sources.source1.type = exec agent1.sources.source1.command = tail -F /home/hadoop/logs/access_log agent1.sources.source1.channels = channel1#configure host for source agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = host agent1.sources.source1.interceptors.i1.hostHeader = hostname# Describe sink1 agent1.sinks.sink1.type = hdfs #a1.sinks.k1.channel = c1 agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = access_log agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text agent1.sinks.sink1.hdfs.rollSize = 102400 agent1.sinks.sink1.hdfs.rollCount = 1000000 agent1.sinks.sink1.hdfs.rollInterval = 60 agent1.sinks.sink1.hdfs.round = true agent1.sinks.sink1.hdfs.roundValue = 10 agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true# Use a channel which buffers events in memory agent1.channels.channel1.type = memory agent1.channels.channel1.keep-alive = 120 agent1.channels.channel1.capacity = 500000 agent1.channels.channel1.transactionCapacity = 600# Bind the source and sink to the channel agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel13、兩個agent級聯
1.3 更多source和sink組件
Flume支持眾多的source和sink類型，詳細手冊可參考官方文檔?http://flume.apache.org/FlumeUserGuide.html
2. sqoop數據遷移工具
2.1 概述
sqoop是apache旗下一款“Hadoop和關系數據庫服務器之間傳送數據”的工具。
導入數據：MySQL，Oracle導入數據到Hadoop的HDFS、HIVE、HBASE等數據存儲系統；
導出數據：從Hadoop的文件系統中導出數據到關系數據庫mysql等
2.2 工作機制
將導入或導出命令翻譯成mapreduce程序來實現，在翻譯出的mapreduce中主要是對inputformat和outputformat進行定制。
2.3 sqoop實戰及原理
2.3.1 sqoop安裝
安裝sqoop的前提是已經具備java和hadoop的環境
1、下載并解壓
最新版下載地址http://ftp.wayne.edu/apache/sqoop/1.4.6/
2、修改配置文件
$ cd $SQOOP_HOME/conf
$ mv sqoop-env-template.sh sqoop-env.sh
打開sqoop-env.sh并編輯下面幾行：
export HADOOP_COMMON_HOME=/home/hadoop/apps/hadoop-2.6.1/ export HADOOP_MAPRED_HOME=/home/hadoop/apps/hadoop-2.6.1/ export HIVE_HOME=/home/hadoop/apps/hive-1.2.13、加入mysql的jdbc驅動包
cp ?~/app/hive/lib/mysql-connector-java-5.1.28.jar ??$SQOOP_HOME/lib/
4、驗證啟動
$ cd $SQOOP_HOME/bin
$ sqoop-version
預期的輸出：
15/12/17 14:52:32 INFO sqoop.Sqoop: Running Sqoop version: 1.4.6
Sqoop 1.4.6?git commit id 5b34accaca7de251fc91161733f906af2eddbe83
Compiled by abe on Fri Aug 1 11:19:26 PDT 2015
到這里，整個Sqoop安裝工作完成。
?
驗證sqoop到mysql業務庫之間的連通性：
bin/sqoop-list-databases --connect jdbc:mysql://localhost:3306 --username root --password root
bin/sqoop-list-tables --connect jdbc:mysql://localhost:3306/userdb?--username root --password root
2.4 Sqoop的數據導入
“導入工具”導入單個表從RDBMS到HDFS。表中的每一行被視為HDFS的記錄。所有記錄都存儲為文本文件的文本數據（或者Avro、sequence文件等二進制數據）?
2.4.1 語法
下面的語法用于將數據導入HDFS
$ sqoop import (generic-args) (import-args)2.4.2 示例
表數據
在mysql中有一個庫userdb中三個表：emp,?emp_add和emp_conn
表emp:
| id | name | deg | salary | dept |
| 1201 | gopal | manager | 50,000 | TP |
| 1202 | manisha | Proof reader | 50,000 | TP |
| 1203 | khalil | php dev | 30,000 | AC |
| 1204 | prasanth | php dev | 30,000 | AC |
| 1205 | kranthi | admin | 20,000 | TP |
表emp_add:
| id | hno | street | city |
| 1201 | 288A | vgiri | jublee |
| 1202 | 108I | aoc | sec-bad |
| 1203 | 144Z | pgutta | hyd |
| 1204 | 78B | old city | sec-bad |
| 1205 | 720X | hitec | sec-bad |
表emp_conn:
| id | phno | |
| 1201 | 2356742 | gopal@tp.com |
| 1202 | 1661663 | manisha@tp.com |
| 1203 | 8887776 | khalil@ac.com |
| 1204 | 9988774 | prasanth@ac.com |
| 1205 | 1231231 | kranthi@tp.com |
導入表表數據到HDFS
下面的命令用于從MySQL數據庫服務器中的emp表導入HDFS
如果成功執行，那么會得到下面的輸出
14/12/22 15:24:54 INFO sqoop.Sqoop: Running Sqoop version: 1.4.5 14/12/22 15:24:56 INFO manager.MySQLManager: Preparing to use a MySQL streaming resultset. INFO orm.CompilationManager: Writing jar file: /tmp/sqoop-hadoop/compile/cebe706d23ebb1fd99c1f063ad51ebd7/emp.jar ----------------------------------------------------- O mapreduce.Job: map 0% reduce 0% 14/12/22 15:28:08 INFO mapreduce.Job: map 100% reduce 0% 14/12/22 15:28:16 INFO mapreduce.Job: Job job_1419242001831_0001 completed successfully ----------------------------------------------------- ----------------------------------------------------- 14/12/22 15:28:17 INFO mapreduce.ImportJobBase: Transferred 145 bytes in 177.5849 seconds (0.8165 bytes/sec) 14/12/22 15:28:17 INFO mapreduce.ImportJobBase: Retrieved 5 records.為了驗證在HDFS導入的數據，請使用以下命令查看導入的數據
$ $HADOOP_HOME/bin/hadoop fs -cat /user/hadoop/emp/part-m-00000emp表的數據和字段之間用逗號(,)表示。
1201, gopal, manager, 50000, TP 1202, manisha, preader, 50000, TP 1203, kalil, php dev, 30000, AC 1204, prasanth, php dev, 30000, AC 1205, kranthi, admin, 20000, TP導入到HDFS指定目錄
在導入表數據到HDFS使用Sqoop導入工具，我們可以指定目標目錄。
以下是指定目標目錄選項的Sqoop導入命令的語法。
--target-dir <new or exist directory in HDFS>下面的命令是用來導入emp_add表數據到'/queryresult'目錄。
bin/sqoop import \ --connect jdbc:mysql://hdp-node-01:3306/test \ --username root \ --password root \ --target-dir /queryresult \ --fields-terminated-by ‘\001’ \ --table emp --split-by id --m 1注意：如果報錯，說emp類找不到，則可以手動從sqoop生成的編譯目錄(/tmp/sqoop-root/compile)中，找到這個emp.class和emp.jar，拷貝到sqoop的lib目錄下：
如果設置了 --m 1，則意味著只會啟動一個maptask執行數據導入
如果不設置 --m 1，則默認為啟動4個map task執行數據導入，則需要指定一個列來作為劃分map task任務的依據
?
下面的命令是用來驗證 /queryresult?目錄中 emp_add表導入的數據形式。
$HADOOP_HOME/bin/hadoop fs -cat /queryresult/part-m-*它會用逗號（，）分隔emp_add表的數據和字段。
1201, 288A, vgiri, jublee 1202, 108I, aoc, sec-bad 1203, 144Z, pgutta, hyd 1204, 78B, oldcity, sec-bad 1205, 720C, hitech, sec-bad導入關系表到HIVE
bin/sqoop import --connect jdbc:mysql://hdp-node-01:3306/test --username root --password root --table emp --hive-import --split-by id --m 1導入表數據子集
我們可以導入表的使用Sqoop導入工具，"where"子句的一個子集。它執行在各自的數據庫服務器相應的SQL查詢，并將結果存儲在HDFS的目標目錄。
where子句的語法如下。
--where <condition>下面的命令用來導入emp_add表數據的子集。子集查詢檢索員工ID和地址，居住城市為：Secunderabad
bin/sqoop import \ --connect jdbc:mysql://hdp-node-01:3306/test \ --username root \ --password root \ --where "city ='sec-bad'" \ --target-dir /wherequery \ --table emp_add \--m 1按需導入
bin/sqoop import \ --connect jdbc:mysql://hdp-node-01:3306/test \ --username root \ --password root \ --target-dir /wherequery2 \ --query 'select id,name,deg from emp WHERE id>1207 and $CONDITIONS' \ --split-by id \ --fields-terminated-by '\t' \ --m 2下面的命令用來驗證數據從emp_add表導入/wherequery目錄
$HADOOP_HOME/bin/hadoop fs -cat /wherequery/part-m-*它用逗號（，）分隔 emp_add表數據和字段。
1202, 108I, aoc, sec-bad 1204, 78B, oldcity, sec-bad 1205, 720C, hitech, sec-bad增量導入
增量導入是僅導入新添加的表中的行的技術。
sqoop支持兩種增量MySql導入到hive的模式，
一種是append，即通過指定一個遞增的列，比如：
--incremental append ?--check-column num_id --last-value 0
另種是可以根據時間戳，比如：
--incremental lastmodified --check-column created --last-value '2012-02-01 11:0:00'
就是只導入created 比'2012-02-01 11:0:00'更大的數據。
?
1/ append模式
它需要添加‘incremental’, ‘check-column’, 和 ‘last-value’選項來執行增量導入。
下面的語法用于Sqoop導入命令增量選項。
--incremental <mode> --check-column <column name> --last value <last check column value>假設新添加的數據轉換成emp表如下：
1206, satish p, grp des, 20000, GR
下面的命令用于在EMP表執行增量導入。
bin/sqoop import \ --connect jdbc:mysql://hdp-node-01:3306/test \ --username root \ --password root \ --table emp --m 1 \ --incremental append \ --check-column id \ --last-value 1208以下命令用于從emp表導入HDFS?emp/?目錄的數據驗證。
$ $HADOOP_HOME/bin/hadoop fs -cat /user/hadoop/emp/part-m-* 1201, gopal, manager, 50000, TP 1202, manisha, preader, 50000, TP 1203, kalil, php dev, 30000, AC 1204, prasanth, php dev, 30000, AC 1205, kranthi, admin, 20000, TP 1206, satish p, grp des, 20000, GR下面的命令是從表emp 用來查看修改或新添加的行
$HADOOP_HOME/bin/hadoop fs -cat /emp/part-m-*1 1206, satish p, grp des, 20000, GR2.5 Sqoop的數據導出
1/ 將數據從HDFS把文件導出到RDBMS數據庫
導出前，目標表必須存在于目標數據庫中。
語法
以下是export命令語法。
$ sqoop export (generic-args) (export-args)示例
數據是在HDFS?中“EMP/”目錄的emp_data文件中。所述emp_data如下：
1201, gopal, manager, 50000, TP 1202, manisha, preader, 50000, TP 1203, kalil, php dev, 30000, AC 1204, prasanth, php dev, 30000, AC 1205, kranthi, admin, 20000, TP 1206, satish p, grp des, 20000, GR1、首先需要手動創建mysql中的目標表
$ mysql mysql> USE db; mysql> CREATE TABLE employee ( id INT NOT NULL PRIMARY KEY, name VARCHAR(20), deg VARCHAR(20),salary INT,dept VARCHAR(10));2、然后執行導出命令
bin/sqoop export \ --connect jdbc:mysql://hdp-node-01:3306/test \ --username root \ --password root \ --table employee \ --export-dir /user/hadoop/emp/3、驗證表mysql命令行。
mysql>select * from employee; 如果給定的數據存儲成功，那么可以找到數據在如下的employee表。 +------+--------------+-------------+-------------------+--------+ | Id | Name | Designation | Salary | Dept | +------+--------------+-------------+-------------------+--------+ | 1201 | gopal | manager | 50000 | TP | | 1202 | manisha | preader | 50000 | TP | | 1203 | kalil | php dev | 30000 | AC | | 1204 | prasanth | php dev | 30000 | AC | | 1205 | kranthi | admin | 20000 | TP | | 1206 | satish p | grp des | 20000 | GR | +------+--------------+-------------+-------------------+--------+?
轉載于:https://www.cnblogs.com/zhangchao162/p/9896805.html
創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎總結
以上是生活随笔為你收集整理的Hadoop辅助工具——Flume、Sqoop的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 微信小程序+微信公众号开发总结
- 下一篇： php实现异步请求