flumemysql正则_利用Flume将MySQL表数据准实时抽取到HDFS

一、为什么要用到Flume

在以前搭建HAWQ数据仓库实验环境时&＃xff0c;我使用Sqoop抽取从MySQL数据库增量抽取数据到HDFS&＃xff0c;然后用HAWQ的外部表进行访问。这种方式只需要很少量的配置即可完成数据抽取任务&＃xff0c;但缺点同样明显&＃xff0c;那就是实时性。Sqoop使用MapReduce读写数据&＃xff0c;而MapReduce是为了批处理场景设计的&＃xff0c;目标是大吞吐量&＃xff0c;并不太关心低延时问题。就像实验中所做的&＃xff0c;每天定时增量抽取数据一次。

Flume是一个海量日志采集、聚合和传输的系统&＃xff0c;支持在日志系统中定制各类数据发送方&＃xff0c;用于收集数据。同时&＃xff0c;Flume提供对数据进行简单处理&＃xff0c;并写到各种数据接受方的能力。Flume以流方式处理数据&＃xff0c;可作为代理持续运行。当新的数据可用时&＃xff0c;Flume能够立即获取数据并输出至目标&＃xff0c;这样就可以在很大程度上解决实时性问题。

Flume是最初只是一个日志收集器&＃xff0c;但随着flume-ng-sql-source插件的出现&＃xff0c;使得Flume从关系数据库采集数据成为可能。下面简单介绍Flume&＃xff0c;并详细说明如何配置Flume将MySQL表数据准实时抽取到HDFS。

二、Flume简介

1. Flume的概念

Flume是分布式的日志收集系统&＃xff0c;它将各个服务器中的数据收集起来并送到指定的地方去&＃xff0c;比如说送到HDFS&＃xff0c;简单来说flume就是收集日志的&＃xff0c;其架构如图1所示。

图1

2. Event的概念

在这里有必要先介绍一下Flume中event的相关概念&＃xff1a;Flume的核心是把数据从数据源(source)收集过来&＃xff0c;在将收集到的数据送到指定的目的地(sink)。为了保证输送的过程一定成功&＃xff0c;在送到目的地(sink)之前&＃xff0c;会先缓存数据(channel),待数据真正到达目的地(sink)后&＃xff0c;Flume再删除自己缓存的数据。

在整个数据的传输的过程中&＃xff0c;流动的是event&＃xff0c;即事务保证是在event级别进行的。那么什么是event呢&＃xff1f;Event将传输的数据进行封装&＃xff0c;是Flume传输数据的基本单位&＃xff0c;如果是文本文件&＃xff0c;通常是一行记录。Event也是事务的基本单位。Event从source&＃xff0c;流向channel&＃xff0c;再到sink&＃xff0c;本身为一个字节数组&＃xff0c;并可携带headers(头信息)信息。Event代表着一个数据的最小完整单元&＃xff0c;从外部数据源来&＃xff0c;向外部的目的地去。

3. Flume架构介绍

Flume之所以这么神奇&＃xff0c;是源于它自身的一个设计&＃xff0c;这个设计就是agent。Agent本身是一个Java进程&＃xff0c;运行在日志收集节点——所谓日志收集节点就是服务器节点。 Agent里面包含3个核心的组件&＃xff1a;source、channel和sink&＃xff0c;类似生产者、仓库、消费者的架构。

Source&＃xff1a;source组件是专门用来收集数据的&＃xff0c;可以处理各种类型、各种格式的日志数据,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定义。

Channel&＃xff1a;source组件把数据收集来以后&＃xff0c;临时存放在channel中&＃xff0c;即channel组件在agent中是专门用来存放临时数据的——对采集到的数据进行简单的缓存&＃xff0c;可以存放在memory、jdbc、file等等。

Sink&＃xff1a;sink组件是用于把数据发送到目的地的组件&＃xff0c;目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定义。

4. Flume的运行机制

Flume的核心就是一个agent&＃xff0c;这个agent对外有两个进行交互的地方&＃xff0c;一个是接受数据输入的source&＃xff0c;一个是数据输出的sink&＃xff0c;sink负责将数据发送到外部指定的目的地。source接收到数据之后&＃xff0c;将数据发送给channel&＃xff0c;chanel作为一个数据缓冲区会临时存放这些数据&＃xff0c;随后sink会将channel中的数据发送到指定的地方&＃xff0c;例如HDFS等。注意&＃xff1a;只有在sink将channel中的数据成功发送出去之后&＃xff0c;channel才会将临时数据进行删除&＃xff0c;这种机制保证了数据传输的可靠性与安全性。

三、安装Hadoop和Flume

我的实验在HDP 2.5.0上进行&＃xff0c;HDP安装中包含Flume&＃xff0c;只要配置Flume服务即可。HDP的安装步骤参见“HAWQ技术解析(二) —— 安装部署”

四、配置与测试

1. 建立MySQL数据库表

建立测试表并添加数据。

use test;

create table  wlslog

(id         int not null,

time_stamp varchar(40),

category   varchar(40),

type       varchar(40),

servername varchar(40),

code       varchar(40),

msg        varchar(40),

primary key ( id )

);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(1,&＃39;apr-8-2014-7:06:16-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000365&＃39;,&＃39;server state changed to standby&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(2,&＃39;apr-8-2014-7:06:17-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000365&＃39;,&＃39;server state changed to starting&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(3,&＃39;apr-8-2014-7:06:18-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000365&＃39;,&＃39;server state changed to admin&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(4,&＃39;apr-8-2014-7:06:19-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000365&＃39;,&＃39;server state changed to resuming&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(5,&＃39;apr-8-2014-7:06:20-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000361&＃39;,&＃39;started weblogic adminserver&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(6,&＃39;apr-8-2014-7:06:21-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000365&＃39;,&＃39;server state changed to running&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(7,&＃39;apr-8-2014-7:06:22-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000360&＃39;,&＃39;server started in running mode&＃39;);

commit;

2. 建立相关目录与文件

(1)创建本地状态文件

mkdir -p /var/lib/flume

cd /var/lib/flume

touch sql-source.status

chmod -R 777 /var/lib/flume

(2)建立HDFS目标目录

hdfs dfs -mkdir -p /flume/mysql

hdfs dfs -chmod -R 777 /flume/mysql

3. 准备JAR包

从http://book2s.com/java/jar/f/flume-ng-sql-source/download-flume-ng-sql-source-1.3.7.html下载flume-ng-sql-source-1.3.7.jar文件&＃xff0c;并复制到Flume库目录。

cp flume-ng-sql-source-1.3.7.jar /usr/hdp/current/flume-server/lib/

将MySQL JDBC驱动JAR包也复制到Flume库目录。

cp mysql-connector-java-5.1.17.jar /usr/hdp/current/flume-server/lib/mysql-connector-java.jar

4. 建立HAWQ外部表

create external table ext_wlslog

(id         int,

time_stamp varchar(40),

category   varchar(40),

type       varchar(40),

servername varchar(40),

code       varchar(40),

msg        varchar(40)

) location (&＃39;pxf://mycluster/flume/mysql?profile&＃61;hdfstextmulti&＃39;) format &＃39;csv&＃39; (quote&＃61;e&＃39;"&＃39;);

5. 配置Flume

在Ambari -> Flume -> Configs -> flume.conf中配置如下属性&＃xff1a;

agent.channels.ch1.type &＃61; memory

agent.sources.sql-source.channels &＃61; ch1

agent.channels &＃61; ch1

agent.sinks &＃61; HDFS

agent.sources &＃61; sql-source

agent.sources.sql-source.type &＃61; org.keedio.flume.source.SQLSource

agent.sources.sql-source.connection.url &＃61; jdbc:mysql://172.16.1.127:3306/test

agent.sources.sql-source.user &＃61; root

agent.sources.sql-source.password &＃61; 123456

agent.sources.sql-source.table &＃61; wlslog

agent.sources.sql-source.columns.to.select &＃61; *

agent.sources.sql-source.incremental.column.name &＃61; id

agent.sources.sql-source.incremental.value &＃61; 0

agent.sources.sql-source.run.query.delay&＃61;5000

agent.sources.sql-source.status.file.path &＃61; /var/lib/flume

agent.sources.sql-source.status.file.name &＃61; sql-source.status

agent.sinks.HDFS.channel &＃61; ch1

agent.sinks.HDFS.type &＃61; hdfs

agent.sinks.HDFS.hdfs.path &＃61; hdfs://mycluster/flume/mysql

agent.sinks.HDFS.hdfs.fileType &＃61; DataStream

agent.sinks.HDFS.hdfs.writeFormat &＃61; Text

agent.sinks.HDFS.hdfs.rollSize &＃61; 268435456

agent.sinks.HDFS.hdfs.rollInterval &＃61; 0

agent.sinks.HDFS.hdfs.rollCount &＃61; 0

Flume在flume.conf文件中指定Source、Channel和Sink相关的配置&＃xff0c;各属性描述如表1所示。

属性

描述

agent.channels.ch1.type

Agent的channel类型

agent.sources.sql-source.channels

Source对应的channel名称

agent.channels

Channel名称

agent.sinks

Sink名称

agent.sources

Source名称

agent.sources.sql-source.type

Source类型

agent.sources.sql-source.connection.url

数据库URL

agent.sources.sql-source.user

数据库用户名

agent.sources.sql-source.password

数据库密码

agent.sources.sql-source.table

数据库表名

agent.sources.sql-source.columns.to.select

查询的列

agent.sources.sql-source.incremental.column.name

增量列名

agent.sources.sql-source.incremental.value

增量初始值

agent.sources.sql-source.run.query.delay

发起查询的时间间隔&＃xff0c;单位是毫秒

agent.sources.sql-source.status.file.path

状态文件路径

agent.sources.sql-source.status.file.name

状态文件名称

agent.sinks.HDFS.channel

Sink对应的channel名称

agent.sinks.HDFS.type

Sink类型

agent.sinks.HDFS.hdfs.path

Sink路径

agent.sinks.HDFS.hdfs.fileType

流数据的文件类型

agent.sinks.HDFS.hdfs.writeFormat

数据写入格式

agent.sinks.HDFS.hdfs.rollSize

目标文件轮转大小&＃xff0c;单位是字节

agent.sinks.HDFS.hdfs.rollInterval

hdfs sink间隔多长将临时文件滚动成最终目标文件&＃xff0c;单位是秒&＃xff1b;如果设置成0&＃xff0c;则表示不根据时间来滚动文件

agent.sinks.HDFS.hdfs.rollCount

当events数据达到该数量时候&＃xff0c;将临时文件滚动成目标文件&＃xff1b;如果设置成0&＃xff0c;则表示不根据events数据来滚动文件

表1

6. 运行Flume代理

保存上一步的设置&＃xff0c;然后重启Flume服务&＃xff0c;如图2所示。

图2

重启后&＃xff0c;状态文件已经记录了将最新的id值7&＃xff0c;如图3所示。

图3

查看目标路径&＃xff0c;生成了一个临时文件&＃xff0c;其中有7条记录&＃xff0c;如图4所示。

图4

查询HAWQ外部表&＃xff0c;结果也有全部7条数据&＃xff0c;如图5所示。

图5

至此&＃xff0c;初始数据抽取已经完成。

7. 测试准实时增量抽取

在源表中新增id为8、9、10的三条记录。

use test;

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(8,&＃39;apr-8-2014-7:06:22-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000360&＃39;,&＃39;server started in running mode&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(9,&＃39;apr-8-2014-7:06:22-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000360&＃39;,&＃39;server started in running mode&＃39;);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(10,&＃39;apr-8-2014-7:06:22-pm-pdt&＃39;,&＃39;notice&＃39;,&＃39;weblogicserver&＃39;,&＃39;adminserver&＃39;,&＃39;bea-000360&＃39;,&＃39;server started in running mode&＃39;);

commit;

5秒之后查询HAWQ外部表&＃xff0c;从图6可以看到&＃xff0c;已经查询出全部10条数据&＃xff0c;准实时增量抽取成功。

图6

五、方案优缺点

利用Flume采集关系数据库表数据最大的优点是配置简单&＃xff0c;不用编程。相比tungsten-replicator的复杂性&＃xff0c;Flume只要在flume.conf文件中配置source、channel及sink的相关属性&＃xff0c;已经没什么难度了。而与现在很火的canal比较&＃xff0c;虽然不够灵活&＃xff0c;但毕竟一行代码也不用写。再有该方案采用普通SQL轮询的方式实现&＃xff0c;具有通用性&＃xff0c;适用于所有关系库数据源。

这种方案的缺点与其优点一样突出&＃xff0c;主要体现在以下几方面。

在源库上执行了查询&＃xff0c;具有入侵性。

通过轮询的方式实现增量&＃xff0c;只能做到准实时&＃xff0c;而且轮询间隔越短&＃xff0c;对源库的影响越大。

只能识别新增数据&＃xff0c;检测不到删除与更新。

要求源库必须有用于表示增量的字段。

即便有诸多局限&＃xff0c;但用Flume抽取关系库数据的方案还是有一定的价值&＃xff0c;特别是在要求快速部署、简化编程&＃xff0c;又能满足需求的应用场景&＃xff0c;对传统的Sqoop方式也不失为一种有效的补充。

参考&＃xff1a;

Flume架构以及应用介绍

Streaming MySQL Database Table Data to HDFS with Flume

how to read data from oracle using FLUME to kafka broker

https://github.com/keedio/flume-ng-sql-source