Hadoop数据收集与入库系统Flume与Sqoop

shisgmei

　　Hadoop提供了一个中央化的存储系统，其有利于进行集中式的数据分析与数据共享。 Hadoop对存储格式没有要求。可以存储用户访问日志、产品信息以及网页数据等数据。
　　常见的两种数据来源。一种是分散的数据源：机器产生的数据、用户访问日志以及用户购买日志。另一种是传统系统中的数据：传统关系型数据库（MySQL、Oracle）、磁盘阵列以及磁带。

　　Flume由三部分构成。Master负责负责通信及配置管理，是集群的控制器。Collector用于对数据进行聚合。往往会产生一个更大的数据流。然后加载到HDFS上。Agent负责采集数据。其是Flume中产生数据流的地方，同时Agent会将产生的数据传输到Collector.
　　Agent 用于采集数据是数据流产生的地方。通常由source和sink两部分组成 。Source用于获取数据，可从文本文件，syslog，HTTP等获 取数据。 Sink将Source获得的数据进一步传输给后面的Collector。 Flume自带了很多source和sink实现。 syslogTcp(5140) | agentSink("localhost",35853) 意思是通过通信协议获取5140端口数据，然后发送到localhost主机的35853端口。 tail("/etc/services") | agentSink("localhost",35853) 意思是读取文件夹/etc/services下文件，然后发送到localhost主机的35853端口。
　　Collector 用于汇总多个Agent结果 ，将汇总结果导入后端存储系统，比如HDFS，HBase。 Flume自带了很多collector实现 collectorSource(35853) | console  表示将收集结果写到控制台。CollectorSource(35853) | collectorSink("file:///tmp/flume/collected", "syslog") 表示将收集结果写到本地目录。collectorSource(35853) | collectorSink("hdfs://namenode/user/flume/ ","syslog"); 表示将收集结果写到hdfs目录。

　　一般为了防止一个collector挂掉所有agent都失效可以将不同的agent连接不同的collector。同时，这样可以保证不同数据类型的agent将数据放到同一个collector。
　　Master 负责管理协调 agent 和collector的配置信息，是Flume集群的控制器。其可跟踪数据流的最后确认信息，并通知agent。 通常需配置多个master以防止单点故障。借助zookeeper管理管理多Master。
　　构建基于Flume的数据收集系统
　　Agent和Collector均可以动态配置。可通过命令行或Web界面配置。 命令行配置 ： 在已经启动的master节点上，依次输入”flume shell””connect localhost ”    如执行 exec config a1 ‘tailDir(“/data/logfile”)’ ‘agentSink’ 。 Web界面 ： 选中节点，填写source、sink等信息。
　　常用架构举例—拓扑1  
　　
　　agentA : tail(“/ngnix/logs”) | agentSink("collector",35853);
　　agentB : tail(“/ngnix/logs”) | agentSink("collector",35853);
　　agentC : tail(“/ngnix/logs”) | agentSink("collector",35853);
　　agentD : tail(“/ngnix/logs”) | agentSink("collector",35853);
　　agentE : tail(“/ngnix/logs”) | agentSink("collector",35853);
　　agentF : tail(“/ngnix/logs”) | agentSink("collector",35853); collector : collectorSource(35853) | collectorSink("hdfs://namenode/flume/","srcdata");

　　agentA : src | agentE2ESink("collectorA",35853);
　　agentB : src | agentE2ESink("collectorA",35853);
　　agentC : src | agentE2ESink("collectorB",35853);
　　agentD : src | agentE2ESink("collectorB",35853);
　　agentE : src | agentE2ESink("collectorC",35853);
　　agentF : src | agentE2ESink("collectorC",35853);
　　collectorA : collectorSource(35853) | collectorSink("hdfs://...","src");
　　collectorB : collectorSource(35853) | collectorSink("hdfs://...","src");
　　collectorC : collectorSource(35853) | collectorSink("hdfs://...","src");

　　agentA : src | agentE2EChain("collectorA:35853","collectorB:35853");
　　agentB : src | agentE2EChain("collectorA:35853","collectorC:35853");
　　agentC : src | agentE2EChain("collectorB:35853","collectorA:35853");
　　agentD : src | agentE2EChain("collectorB:35853","collectorC:35853");
　　agentE : src | agentE2EChain("collectorC:35853","collectorA:35853");
　　agentF : src | agentE2EChain("collectorC:35853","collectorB:35853");
　　collectorA : collectorSource(35853) | collectorSink("hdfs://...","src");
　　collectorB : collectorSource(35853) | collectorSink("hdfs://...","src");
　　collectorC : collectorSource(35853) | collectorSink("hdfs://...","src");
　　传统数据库与Hadoop间数据同步
　　Sqoop：SQL-to-Hadoop，是连接传统关系型数据库和Hadoop 的桥梁。可以把关系型数据库的数据导入到 Hadoop 系统 ( 如 HDFS HBase 和 Hive) 中或把数据从 Hadoop 系统里抽取并导出到关系型数据库里/利用MapReduce可以加快数据传输速度实现批处理方式进行数据传输。
　　Sqoop优势是高效、可控地利用资源、任务并行度，超时时间等 、数据类型映射与转换 可自动进行或用户也可自定义 。同时，其 支持多种数据库如：MySQL 、Oracle 以及PostgreSQL 。

　　通过命令sqoop可以生成map task将传统数据库中的数据传输到HDFS、Hbase或Hive上。
　　Sqoop import 将数据从关系型数据库导入Hadoop中
　　
　　
　　步骤1：Sqoop与数据库Server通信，获取数据库表的元数据 信息； 步骤2：Sqoop启动一个MapOnly的MR作业，利用元数据信 息并行将数据写入Hadoop。
　　Sqoop import使用：
　　sqoop import \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \   
　　--username sqoop \     
　　--password sqoop \     
　　--table cities
　　--connnect: 指定JDBC URL
　　--username/password：mysql数据库的用户名
　　--table：要读取的数据库表
　　bin/hadoop fs -cat cities/part-m-*

　　1,USA,Palo>　　2,Czech Republic,Brno
　　3,USA,Sunnyvale
　　导入到指定目录：
　　sqoop import \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \   
　　--password sqoop \     
　　--table cities \     
　　--target-dir /etl/input/cities
　　导入字段country为USA的值：
　　sqoop import \   
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     
　　--where "country = 'USA'"
　　生成顺序文件：
　　sqoop import \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     
　　--as-sequencefile
　　指定map个数：
　　sqoop import \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     
　　--num-mappers 10
　　Sqoop import—导入多个表 ：
　　sqoop import \   
　　--connect jdbc:mysql://mysql.example.com/sqoop \  
　　--username sqoop \   
　　--password sqoop \   
　　--query 'SELECT normcities.id, \                  
　　countries.country, \                  
　　normcities.city \                  
　　FROM normcities \                  
　　JOIN countries USING(country_id) \                  
　　WHERE $CONDITIONS' \

　　--split-by>　　--target-dir cities
　　Sqoop import增量导入（一）：
　　适用于数据每次被追加到数据库中，而已有数据不变的情况且仅导入id这一列值大于1的记录。
　　sqoop import \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \   
　　--password sqoop \     
　　--table visits \     
　　--incremental append \     

　　--check-column>　　--last-value 1
　　Sqoop import增量导入（二）
　　每次成功运行后，sqoop将最后一条记录的id值保存到 metastore中，供下次使用。
　　sqoop job \   
　　--create visits \   
　　--import \   
　　--connect jdbc:mysql://mysql.example.com/sqoop \   
　　--username sqoop \   
　　--password sqoop \   
　　--table visits \   
　　--incremental append \   

　　--check-column>　　--last-value 0
　　运行sqoop作业：sqoop job --exec visits
　　Sqoop import增量导入（三）
　　数据库中有一列last_update_date，记录了上次修改时间。 Sqoop仅将某时刻后的数据导入Hadoop。
　　sqoop import \   
　　--connect jdbc:mysql://mysql.example.com/sqoop \   
　　--username sqoop \   
　　--password sqoop \   
　　--table visits \   
　　--incremental lastmodified \   
　　--check-column last_update_date \   
　　--last-value “2013-05-22 01:01:01”
　　Sqoop Export 将数据从Hadoop导入关系型数据库导中
　　
　　步骤1：Sqoop与数据库Server通信，获取数据库表的元数据 信息； 步骤2：并行导入数据：将Hadoop上文件划分成若 干个split； 每个split由一个Map Task进 行数据导入。
　　Sqoop Export使用方法
　　sqoop export \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     
　　--export-dir cities
　　--connnect: 指定JDBC URL
　　--username/password：mysql数据库的用户名
　　--table：要导入的数据库表
　　export-dir：数据在HDFS上存放目录
　　Sqoop Export—保证原子性：
　　sqoop export \   
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     
　　--staging-table staging_cities
　　Sqoop Export—更新已有数据：
　　sqoop export \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     

　　--update-key>　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \     
　　--password sqoop \     
　　--table cities \     

　　--update-key>　　--update-mode allowinsert
　　Sqoop Export—选择性插入 ：
　　sqoop export \     
　　--connect jdbc:mysql://mysql.example.com/sqoop \     
　　--username sqoop \   
　　--password sqoop \     
　　--table cities \     
　　--columns country,city
　　Sqoop与其他系统结合
     Sqoop可以与Oozie、Hive、Hbase等系统结合；  用户需要在sqoop-env.sh中增加HBASE_HOME、 HIVE_HOME等环境变量。 Sqoop与Hive结合：　　sqoop import \     
--connect jdbc:mysql://mysql.example.com/sqoop \     --username sqoop \     --password sqoop \     --table cities \     --hive-import　　Sqoop与HBase结合：
　　sqoop import \     
--connect jdbc:mysql://mysql.example.com/sqoop \     --username sqoop \     --password sqoop \     --table cities \     --hbase-table cities \     --column-family world       以上介绍了数据收集系统以及数据传输工具sqoop，后续将讲解Hive及Pig相关主题。来源： http://lib.csdn.net/article/hadoop/62796