文档章节

Flume连接oracle实时推送数据到kafka

四叶草666
 四叶草666
发布于 2017/07/20 09:18
字数 647
阅读 12
收藏 0
点赞 0
评论 0

版本号:

RedHat6.5   JDK1.8    flume-1.6.0   kafka_2.11-0.8.2.1

flume安装

RedHat6.5安装单机flume1.6:RedHat6.5安装单机flume1.6

kafka安装

RedHat6.5安装kafka集群 : RedHat6.5安装kafka集群

1、下载flume-ng-sql-source-1.4.3.jar

CSDN下载地址:http://download.csdn.net/detail/chongxin1/9892184

flume-ng-sql-source-1.4.3.jar是flume用于连接数据库的重要支撑jar包。

2、把flume-ng-sql-source-1.4.3.jar放到flume的lib目录下

 

3、把oracle(此处用的是oracle库)的驱动包放到flume的lib目录下

oracle的jdbc驱动包,放在oracle安装目录下,路径为:D:\app\product\11.2.0\dbhome_1\jdbc\lib

如图:

把ojdbc5.jar放到flume的lib目录下,如图:

4、新建flume-sql.conf

在conf目录新建flume-sql.conf :

 
  1. touch /usr/local/flume/apache-flume-1.6.0-bin/conf/flume-sql.conf
  2. sudo gedit /usr/local/flume/apache-flume-1.6.0-bin/conf/flume-sql.conf

flume-sql.conf输入以下内容:

 

 
  1. agentOne.channels = channelOne
  2. agentOne.sources = sourceOne
  3. agentOne.sinks = sinkOne
  4. ###########sql source#################
  5. # For each one of the sources, the type is defined
  6. agentOne.sources.sourceOne.type = org.keedio.flume.source.SQLSource
  7. agentOne.sources.sourceOne.hibernate.connection.url = jdbc:oracle:thin:@192.168.168.100:1521/orcl
  8. # Hibernate Database connection properties
  9. agentOne.sources.sourceOne.hibernate.connection.user = flume
  10. agentOne.sources.sourceOne.hibernate.connection.password = 1234
  11. agentOne.sources.sourceOne.hibernate.connection.autocommit = true
  12. agentOne.sources.sourceOne.hibernate.dialect = org.hibernate.dialect.Oracle10gDialect
  13. agentOne.sources.sourceOne.hibernate.connection.driver_class = oracle.jdbc.driver.OracleDriver
  14. agentOne.sources.sourceOne.run.query.delay=10000
  15. agentOne.sources.sourceOne.status.file.path = /tmp
  16. agentOne.sources.sourceOne.status.file.name = sqlSource.status
  17. # Custom query
  18. agentOne.sources.sourceOne.start.from = 0
  19. agentOne.sources.sourceOne.custom.query = select sysdate from dual
  20. agentOne.sources.sourceOne.batch.size = 1000
  21. agentOne.sources.sourceOne.max.rows = 1000
  22. agentOne.sources.sourceOne.hibernate.connection.provider_class = org.hibernate.connection.C3P0ConnectionProvider
  23. agentOne.sources.sourceOne.hibernate.c3p0.min_size=1
  24. agentOne.sources.sourceOne.hibernate.c3p0.max_size=10
  25. ##############################
  26. agentOne.channels.channelOne.type = memory
  27. agentOne.channels.channelOne.capacity = 10000
  28. agentOne.channels.channelOne.transactionCapacity = 10000
  29. agentOne.channels.channelOne.byteCapacityBufferPercentage = 20
  30. agentOne.channels.channelOne.byteCapacity = 800000
  31.  
  32. agentOne.sinks.sinkOne.type = org.apache.flume.sink.kafka.KafkaSink
  33. agentOne.sinks.sinkOne.topic = test
  34. agentOne.sinks.sinkOne.brokerList = 192.168.168.200:9092
  35. agentOne.sinks.sinkOne.requiredAcks = 1
  36. agentOne.sinks.sinkOne.batchSize = 20
  37. agentOne.sinks.sinkOne.channel = channelOne
  38.  
  39. agentOne.sinks.sinkOne.channel = channelOne
  40. agentOne.sources.sourceOne.channels=channelOne

5、flume-ng启动flume-sql.conf和测试

 

 
  1. cd /usr/local/flume/apache-flume-1.6.0-bin
  2. bin/flume-ng agent --conf conf --conf-file conf/flume-sql.conf --name agentOne -Dflume.root.logger=INFO,console

运行成功日志如下:

 
  1. 2017-07-08 00:12:55,393 (lifecycleSupervisor-1-1) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)] Monitored counter group for type: SINK, name: sinkOne: Successfully registered new MBean.
  2. 2017-07-08 00:12:55,394 (lifecycleSupervisor-1-1) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)] Component type: SINK, name: sinkOne started
  3. 2017-07-08 00:12:55,463 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Fetching metadata from broker id:0,host:localhost,port:9092 with correlation id 0 for 1 topic(s) Set(test)
  4. 2017-07-08 00:12:55,528 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Connected to localhost:9092 for producing
  5. 2017-07-08 00:12:55,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Disconnecting from localhost:9092
  6. 2017-07-08 00:12:55,582 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Connected to slave2:9092 for producing

启动kafka的消费者,监听topic主题:

 
  1. kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

运行成功日志如下:

 
  1. [root@master kafka_2.11-0.9.0.0]# kafka-console-consumer.sh --zookeeper localhost:2181 --topic test  
  2. "2017-07-08 00:28:53.0"
  3. "2017-07-08 00:29:03.0"
  4. "2017-07-08 00:29:13.0"
  5. "2017-07-08 00:29:23.0"
  6. "2017-07-08 00:29:33.0"
  7. "2017-07-08 00:29:43.0"
  8. "2017-07-08 00:29:53.0"
  9. "2017-07-08 00:30:03.0"

 

6、常见报错解决办法

 
  1.  2017-06-27 16:26:01,293 (C3P0PooledConnectionPoolManager[identityToken->1hgey889o1sjxqn51anc3fr|29938ba5]-AdminTaskTimer) [WARN - com.mchange.v2.async.ThreadPoolAsynchronousRunner$DeadlockDetector.run(ThreadPoolAsynchronousRunner.java:759)] com.mchange.v2.async.ThreadPoolAsynchronousRunner$DeadlockDetector@2d6227f3 -- APPARENT DEADLOCK!!! Complete Status:​

连接超时,造成死锁,仔细检查jdbc:oracle:thin:@192.168.168.100:1521/orcl,用户名/密码是否正确;

如果正确,还是连接不上,请检查oralce数据库是否开启了防火墙,如果是,添加入站规则或直接关闭防火墙。

© 著作权归作者所有

共有 人打赏支持
四叶草666
粉丝 0
博文 51
码字总数 50778
作品 0
深圳
程序员
Kafka实战-Flume到Kafka

1.概述   前面给大家介绍了整个Kafka项目的开发流程,今天给大家分享Kafka如何获取数据源,即Kafka生产数据。下面是今天要分享的目录: 数据来源 Flume到Kafka 数据源加载 预览   下面开...

smartloli ⋅ 2015/07/02 ⋅ 0

【Strom篇】Flume+Kafaka+Strom整合完成信息记录

第一部分:流程分析 由flume收集客户端发送的信息,经过kafka集群消费者收集,然后给storm集群数据处理(数据清洗),最后再发给kafka集群收集。 第二部分:集群搭建 flume+kafka 一、配合f...

chenshi_2753 ⋅ 05/06 ⋅ 0

kafka来读取flume的数据

一、查看kafka topic ./kafka-topics.sh --list --zookeeper bigdata-test-3:2181, bigdata-test-2:2181, bigdata-test-1:2181, bigdata-test-4:2181, bigdata-test-5:2181 ./kafka-topics.s......

weixin_41876523 ⋅ 05/24 ⋅ 0

Flume+Kafka+Storm+Redis构建大数据实时处理系统:实时统计网站PV、UV+展示

[TOC] 1 大数据处理的常用方法 前面在我的另一篇文章中《大数据采集、清洗、处理:使用MapReduce进行离线数据分析完整案例》中已经有提及到,这里依然给出下面的图示: 前面给出的那篇文章是...

xpleaf ⋅ 04/16 ⋅ 0

Flume整合Kafka实时收集日志信息

Linux系统查看文件内容的特殊方法: 最基本的有cat和less,more,如果有特殊的要求的话。 1/如果只想看文件的前5行,可以使用命令,如: 2/如果想查看文件的后10行,可以使用命令,如: 3/参数...

sparkle123 ⋅ 05/10 ⋅ 0

Flume---大数据协作框架

flume是什么 Apache Flume是一个分布式的、可靠的、易用的系统,可以有效地将来自很多不同源系统的大量日志数据收集、汇总或者转移到一个数据中心存储。 Apache Flume的作用不仅限于日志汇总...

简心 ⋅ 05/06 ⋅ 0

flume 总结--flume入门介绍

flume介绍 flume被设计为一个灵活的分布式系统,可以很容易的扩展,而且是高度可定制化的,一个配置正确的Flume Agent和由互相连接的Agent创建的Agent管道,保证不会丢失数据,提供持久的cha...

u013362353 ⋅ 05/28 ⋅ 0

其他消息中间件及场景应用(下3)

版权声明:欢迎转载,但是看在我辛勤劳动的份上,请注明来源:http://blog.csdn.net/yinwenjie(未经允许严禁用于商业用途!) https://blog.csdn.net/yinwenjie/article/details/51516329 目...

yunlielai ⋅ 04/15 ⋅ 0

解决Flume采集数据时在HDFS上产生大量小文件的问题

问题:flume指定HDFS类型的Sink时,采集数据至HDFS指定目录,会产生大量小文件。 问题重现: 1、创建flume配置文件flume-env.sh,: flume配置文件如下(根据自身需要修改): 因为flume可以...

舒运 ⋅ 06/10 ⋅ 0

flume_kafkaChannel_kafkaSink

agent.sources = source 抽取类型为目录 agent.sources.source.type = spooldir 抽取的文件目录 agent.sources.source.spoolDir = /root/tmp/flume/data 添加一个存储绝对路径文件名的头 ag...

tanj123 ⋅ 04/17 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

BS与CS的联系与区别【简】

C/S是Client/Server的缩写。服务器通常采用高性能的PC、工作站或小型机,并采用大型数据库系统,如Oracle、Sybase、InFORMix或 SQL Server。客户端需要安装专用的客户端软件。 B/S是Brower/...

anlve ⋅ 34分钟前 ⋅ 0

发生了什么?Linus 又发怒了?

在一个 Linux 内核 4.18-rc1 的 Pull Request 中,开发者 Andy Shevchenko 表示其在对设备属性框架进行更新时,移除了 union 别名,这引发了 Linus 的暴怒。 这一次 Linus Torvalds 发怒的原...

问题终结者 ⋅ 54分钟前 ⋅ 0

在树莓派上搭建一个maven仓库

在树莓派上搭建一个maven仓库 20180618 lambo init 项目说明 家里有台树莓派性能太慢。想搭建一个maven私服, 使用nexus或者 jfrog-artifactory 运行的够呛。怎么办呢,手写一个吧.所在这个...

林小宝 ⋅ 今天 ⋅ 0

Spring发展历程总结

转自与 https://www.cnblogs.com/RunForLove/p/4641672.html 目前很多公司的架构,从Struts2迁移到了SpringMVC。你有想过为什么不使用Servlet+JSP来构建Java web项目,而是采用SpringMVC呢?...

onedotdot ⋅ 今天 ⋅ 0

Python模块/包/库安装(6种方法)

Python模块/包/库安装(6种方法) 冰颖机器人 2016-11-29 21:33:26 一、方法1: 单文件模块 直接把文件拷贝到 $python_dir/Lib 二、方法2: 多文件模块,带setup.py 下载模块包(压缩文件zip...

cswangyx ⋅ 今天 ⋅ 0

零基础学习大数据人工智能,学习路线篇!系统规划大数据之路?

大数据处理技术怎么学习呢?首先我们要学习Python语言和Linux操作系统,这两个是学习大数据的基础,学习的顺序不分前后。 Python:Python 的排名从去年开始就借助人工智能持续上升,现在它已经...

董黎明 ⋅ 今天 ⋅ 0

openJdk和sun jdk的区别

使用过LINUX的人都应该知道,在大多数LINUX发行版本里,内置或者通过软件源安装JDK的话,都是安装的OpenJDK, 那么到底什么是OpenJDK,它与SUN JDK有什么关系和区别呢? 历史上的原因是,Ope...

jason_kiss ⋅ 今天 ⋅ 0

梳理

Redux 是 JavaScript 状态容器,提供可预测化的状态管理。 它是JS的状态容器,是一种解决问题的方式,所以即可以用于 react 也可以用于 vue。 需要理解其思想及实现方式。 应用中所有的 stat...

分秒 ⋅ 今天 ⋅ 0

Java 后台判断是否为ajax请求

/** * 是否是Ajax请求 * @param request * @return */public static boolean isAjax(ServletRequest request){return "XMLHttpRequest".equalsIgnoreCase(((HttpServletReques......

JavaSon712 ⋅ 今天 ⋅ 0

Redis 单线程 为何却需要事务处理并发问题

Redis是单线程处理,也就是命令会顺序执行。那么为什么会存在并发问题呢? 个人理解是,虽然redis是单线程,但是可以同时有多个客户端访问,每个客户端会有 一个线程。客户端访问之间存在竞争...

码代码的小司机 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部