flume全量抽取oracle,flume监控Oracle表变化增量抽取数据到kafka

1.涉及到的软件框架及版本号

系统及软件

版本

CentOS

centos7.2

JDK

JDK1.8

Flume

apache-flume-1.9.0-bin.tar.gz

kafka

kafka_2.11-0.11.0.2.tgz

zookeper

zookeeper-3.4.10.tar.gz

oracle

2.zookeper单机版安装

1)下载安装zookeeper，并配置环境变量

2)修改配置文件 zookeeper/conf/zoo.cfg (zoo_sample.cfg --> zoo.cfg)

dataDir=xxxx

3)启动zk:

bin/zkServer.sh start

4)查看zk状态:

bin/zkServer.sh status

5)停用zk:

bin/zkServer.sh stop

3.kafka单机版安装

1)下载安装kafka，配置环境变量

2)修改配置文件 kafka/config/server.properties

broker.id=0 listeners=PLAINTEXT://:9092 host.name=localhost log.dirs=/Users/Mac/app/kafka_2.11-0.9.0.0/tmp/kafka-logs zookeeper.connect=localhost:2181

3)启动 kafka

kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

4)创建 topic

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_topic

5)启动一个producer发送信息

kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic

6)启动一个consumer接收信息

kafka-console-consumer.sh --zookeeper localhost:2181 --topic test_topic

==> 添加 --from-beginning：表示从头开始消费，不加则只会消费新消息

7)查看创建的topic的信息

kafka-topics.sh --describe --zookeeper localhost:2181

==> 添加 --topic test_topic，查看指定topic的信息

8)查看所有已经创建的topic

kafka-topics.sh --list --zookeeper localhost:2181

9)关闭kafka

kafka-server-stop.sh

4.flume安装

解压flume

tar -zxvf apache-flume-1.9.0-bin.tar.gz

5.flume-ng-sql-source安装

flume默认是不支持sql source的，因此需要先添加flume-ng-sql-source  支持MySql SQLServer DB2

github地址为https://github.com/keedio/flume-ng-sql-source 根据自己的环境选择版本，版本要求：flume-1.7.0以下的版本最好是使用1.4.3，稳定。flume-1.8的话稳定版应该是1.5.0。README中也有基本配置说明。

打包前注意将pom文件中的 flume-ng-core 版本与flume安装版本保持一致，不然运行会报错。

(1)Git Clone代码到本地:

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

(2)切换标签选择自己需要的版本

git checkout 1.5.0

(3)打开pom.xml文件改变相对应的flue版本

org.apache.flume

flume-ng-core

1.9.0

provided

(4)插件编译和封装(最好配置好flume环境变量)

mvn clean package -Dmaven.test.skip=true

编译好后在target文件下有个flume-ng-sql-source-1.5.0.jar包

(5) 扩展部署

mkdir -p $FLUME_HOME/plugins.d/sql-source/lib

mkdir -p $FLUME_HOME/plugins.d/sql-source/libext

cp flume-ng-sql-source-1.5.0.jar $FLUME_HOME/plugins.d/sql-source/lib

(6)数据库引擎的特定安装

把oracle(此处用的是oracle库)的驱动包ojdbc5.jar放到上面创建的目录下

oracle的jdbc驱动包，放在oracle安装目录下，路径为：D:appproduct11.2.0dbhome_1jdbclib下的ojdbc5.jar放到libext目录下

cp ojdbc5.jar $FLUME_HOME/plugins.d/sql-source/libext

(7)新建flume-sql.conf

在conf目录新建flume-sql.conf 输入以下内容：

a1.channels = c1

a1.sources = r1

a1.sinks = k1

###########sql source#################

# For each one of the sources, the type is defined

#source组件类型名称，必须为org.keedio.flume.source.SQLSource

a1.sources.r1.type = org.keedio.flume.source.SQLSource

#与远程数据库连接的网址,这里我是oracle

a1.sources.r1.hibernate.connection.url = jdbc:oracle:thin:@192.168.100.235:1521/cc

# Hibernate Database connection properties

#用于连接数据库的用户名

a1.sources.r1.hibernate.connection.user = system

#连接数据库的密码

a1.sources.r1.hibernate.connection.password = manager

#是否自动提交

a1.sources.r1.hibernate.connection.autocommit = true

#Dialect to use by hibernate

a1.sources.r1.hibernate.dialect = org.hibernate.dialect.Oracle10gDialect

#驱动类

a1.sources.r1.hibernate.connection.driver_class = oracle.jdbc.driver.OracleDriver

#在运行查询完成后等待时间(ms)

a1.sources.r1.run.query.delay=500

#保存状态文件的路径

a1.sources.r1.status.file.path = /usr/local/modul/bigdata/apache-flume-1.9.0-bin/flume/sqlSource

#保存状态文件的名称

a1.sources.r1.status.file.name = flume.sqlSource.status

# Custom query

#查询开始的标识，使用你增量的字段。这里我用的时间毫秒值来做增量判断，可以根据业务来更改

#2017-01-01 08:00:00

a1.sources.r1.start.from = 1483228800000

#自定义查询语句，一定要确保查询结果的第一位置返回增量字段

a1.sources.r1.custom.query = select * from (select TO_NUMBER (TO_DATE (TO_CHAR (a.create_time,'YYYY/MM/DD HH24:MI:SS'),'YYYY/MM/DD HH24:MI:SS') - TO_DATE ('1970/01/01 0:0:0','YYYY/MM/DD HH24:MI:SS')) * 24 * 60 * 60 * 1000 date_number,a.* from MEDREC.event_task a where a.event_type='1' and a.service_code like '%EMPI%' and rownum < 100) t where t.date_number > $@$ ORDER BY t.date_number ASC

#批量大小

a1.sources.r1.batch.size = 1000

#最大行数

a1.sources.r1.max.rows = 1000

#连接池驱动类

a1.sources.r1.hibernate.connection.provider_class = org.hibernate.connection.C3P0ConnectionProvider

#连接池最小个数

a1.sources.r1.hibernate.c3p0.min_size=1

#连接池最大个数

a1.sources.r1.hibernate.c3p0.max_size=10

##############################

#channel的类型

a1.channels.c1.type = memory

#在channel中最多能保存多少个event

a1.channels.c1.capacity = 10000

#在每次从source获取数据后者将数据sink出去的一次事务操作中，最多处理的event数

a1.channels.c1.transactionCapacity = 10000

#在channel中最多容纳所有event body的总字节数

a1.channels.c1.byteCapacity = 800000

#这个值得含义跟上面一样，只不过这个计算event header跟最大可用内存的字节占比

a1.channels.c1.byteCapacityBufferPercentage = 20

##############################

#sink的类型

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink

#kafka topic

a1.sinks.k1.topic = test_topic

#brokerList

a1.sinks.k1.kafka.bootstrap.servers = 192.168.101.32:9092

#ack机制：1表示producer只要收到一个leader分区副本成功写入的通知就认为推送消息成功了。0代表producer发送一次就不在发送，不管成不成功。-1代表producer只有收到分区内所有副本成功写入的通知才认为推送成功

a1.sinks.k1.producer.acks = 1

#batch的大小

a1.sinks.k1.flumeBatchSize = 20

a1.sinks.k1.channel = c1

a1.sinks.k1.channel = c1

a1.sources.r1.channels=c1

(8)flume-ng启动flume-sql.conf和测试

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

(9)启动kafka的消费者，监听topic主题

kafka-console-consumer.sh --zookeeper localhost:2181 --topic test_topic