前言
最近在利用Spark streaming
和Kafka
构建一个实时的数据分析系统,对图书阅读数据进行分析,做实时推荐。Spark Streaming 模块是对于 Spark Core 的一个扩展,目的是为了以高吞吐量,并且容错的方式处理持续性的数据流。目前 Spark Streaming 支持的外部数据源有 Flume、 Kafka、Twitter、ZeroMQ、TCP Socket 等。Apache Kafka
是一个分布式的消息发布-订阅系统,Kafka可以作为流计算系统的数据源,本例中Spark Streaming
将从Kafka
中消费数据。
系统环境
软件版本
1 2 3 |
Spark: 1.4.1 Kafka: 0.8.1.1 zookeeper: 3.4.6 |
集群节点
一共有四台主机,主机名分别为nn0001, dn0001, dn0002, dn0003。
1 2 3 4 |
192.168.186.12 nn0001 192.168.186.13 dn0001 192.168.186.14 dn0002 192.168.186.15 dn0003 |
zookeeper安装
kafka使用zookeeper来管理,存储一些meta信息,并使用了zookeeper watch机制来发现meta信息的变更并作出相应的动作(比如consumer失效,触发负载均衡等)。 Zookeeper的配置在机器1上完成后分发到其他三台机器即可。
1 2 3 4 5 6 |
[bigdata@nn0001 ~]$ wget http://archive.apache.org/dist/zookeeper/stable/zookeeper-3.4.6.tar.gz [bigdata@nn0001 ~]$ tar -zxvf zookeeper-3.4.6.tar.gz [bigdata@nn0001 ~]$cd zookeeper-3.4.6/conf [bigdata@nn0001 conf]$ pwd /home/bigdata/bigprosoft/zookeeper-3.4.6/conf [bigdata@nn0001 conf]$ cp zoo_sample.cfg zoo.cfg |
修改配置文件
1 2 3 4 5 6 7 8 9 10 |
[bigdata@nn0001 conf]$ vi zoo.cfg tickTime=2000 dataDir=/home/bigdata/bigprosoft/zookeeper/data clientPort=2181 initLimit=10 syncLimit=5 server.1=nn0001:2888:3888 server.2=dn0001:2888:3888 server.3=dn0002:2888:3888 server.4=dn0003:2888:3888 |
在dataDir目录下创建myid文件,nn0001机器的内容为1,dn0001机器的内容为2,更多依此类推。
1 2 3 |
[bigdata@nn0001 data]$ echo 1 > myid [bigdata@nn0001 data]$ cat myid 1 |
启动测试
1 2 3 4 5 6 7 8 9 10 |
[bigdata@nn0001 bin]$ ./zkServer.sh start [bigdata@nn0001 bin]$ jps 10805 QuorumPeerMain #已经启动成功了 15494 Master 11816 NameNode 20958 Jps 17539 Worker 12084 ResourceManager 12945 RunJar 12944 RunJar |
停止
1 |
[bigdata@nn0001 bin]$ ./zkServer.sh stop |
其它机器相同操作,scp过去即可。
kafka安装
Kafka的broker、producer、consumer、topic等概念以及原理可以查阅官方文档 本次实验采用的多节点多broker集群模式,为每一台机器分配一个broker id
。
1 2 3 4 5 6 7 8 9 10 |
[bigdata@nn0001 ~]$ wget http://mirror.bit.edu.cn/apache/kafka/0.8.1.1/kafka_2.10-0.8.1.1.tgz [bigdata@nn0001 ~]$ tar zxf kafka_2.10-0.8.1.1.tgz [bigdata@nn0001 ~]$ cd kafka_2.10-0.8.1.1 [bigdata@nn0001 kafka_2.10-0.8.1.1]$ cd conf [bigdata@nn0001 conf]$ vi server.properties broker.id=1 #其它机器的id依次递增即可 port=9092 host.name=192.168.186.12 advertised.host.name=192.168.186.12 zookeeper.connect=192.168.186.12:2181,192.168.186.13:2181,192.168.186.14:2181,192.168.186.15:2181 |
修改完成后分发到另外三台机器上。
启动测试
1 2 3 4 5 6 7 8 9 10 11 |
[bigdata@nn0001 bin]$ nohup ./kafka-server-start.sh ../config/server.properties & [bigdata@nn0001 conf]$ jps 10805 QuorumPeerMain 21282 Jps 15494 Master 21209 Kafka 11816 NameNode 17539 Worker 12084 ResourceManager 12945 RunJar 12944 RunJar |
依次启动机器
kafka使用测试
创建topic
1 |
[bigdata@nn0001 bin]$ ./kafka-topics.sh --create --zookeeper nn0001:2181 --replication-factor 3 --partitions 1 --topic test |
查看topic
1 2 3 4 5 6 |
[bigdata@nn0001 bin]$ ./kafka-topics.sh --describe --zookeeper nn0001:2181 Topic:mytest PartitionCount:2 ReplicationFactor:2 Configs: Topic: mytest Partition: 0 Leader: 2 Replicas: 3,2 Isr: 2 Topic: mytest Partition: 1 Leader: -1 Replicas: 4,3 Isr: Topic:test PartitionCount:1 ReplicationFactor:3 Configs: Topic: test Partition: 0 Leader: 2 Replicas: 2,3,4 Isr: 2 |
producer测试
1 2 3 |
[bigdata@nn0001 bin]$ ./kafka-console-producer.sh --broker-list 192.168.186.12:9092 --topic test gsdggfgfgfd gdfgdfgdf |
conumer测试
1 2 3 4 5 6 7 8 9 10 |
[bigdata@nn0001 bin]$ ./kafka-console-consumer.sh --zookeeper 192.168.186.12:2181 --from-beginning --topic test |
测试高可用
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
[bigdata@nn0001 bin]$ ./kafka-topics.sh --describe --zookeeper 192.168.186.12:2181,192.168.186.13:2181,192.168.186.14:2181,192.168.186.15:2181 --from-beginning --topic test Topic:test PartitionCount:1 ReplicationFactor:3 Configs: Topic: test Partition: 0 Leader: 2 Replicas: 2,3,4 Isr: 2,4 #可以看到leader是2,是dn0001机器,把此机器上的kafka进程杀掉,再查看topic的leader |
ok,搭建过程就完成,下面用python/java/scala进行开发实例即可。
排错
问题1描述
1 2 3 |
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. |
解决方法
1 2 3 |
[bigdata@nn0001 ~]$ wget http://www.slf4j.org/dist/slf4j-1.7.12.tar.gz [bigdata@nn0001 ~]$ cd slf4j-1.7.12 [bigdata@nn0001 ~]$ cp slf4j-nop-1.7.12.jar ~/bigprosoft/kafka/libs/ |
问题2描述
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
[bigdata@nn0001 bin]$ ./kafka-console-producer.sh --broker-list nn0001:9092 --topic test fsfsdfsdf …… [2015-08-28 17:24:18,417] ERROR Failed to send requests for topics test with correlation ids in [0,8] (kafka.producer.async.DefaultEventHandler) [2015-08-28 17:24:18,419] ERROR Error in handling batch of 1 events (kafka.producer.async.ProducerSendThread) kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries. at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67) at scala.collection.immutable.Stream.foreach(Stream.scala:547) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44) …… |
解决方法,把server.properties中主机名改为IP地址即可。
1 2 3 |
host.name=10.171.59.221 advertised.host.name=10.171.59.221 zookeeper.connect=192.168.186.12:2181,192.168.186.13:2181,192.168.186.14:2181,192.168.186.15:2181 |
kafak集群安装-转的更多相关文章
-
kafka2.9.2的伪分布式集群安装和demo(java api)测试
目录: 一.什么是kafka? 二.kafka的官方网站在哪里? 三.在哪里下载?需要哪些组件的支持? 四.如何安装? 五.FAQ 六.扩展阅读 一.什么是kafka? kafka是LinkedI ...
-
ubuntu12.04+kafka2.9.2+zookeeper3.4.5的伪分布式集群安装和demo(java api)测试
博文作者:迦壹 博客地址:http://idoall.org/home.php?mod=space&uid=1&do=blog&id=547 转载声明:可以转载, 但必须以超链 ...
-
【Oracle 集群】Oracle 11G RAC教程之集群安装(七)
Oracle 11G RAC集群安装(七) 概述:写下本文档的初衷和动力,来源于上篇的<oracle基本操作手册>.oracle基本操作手册是作者研一假期对oracle基础知识学习的汇总. ...
-
kafka集群安装部署
kafka集群安装 使用的版本 系统:centos6.5 centos6.7 jdk:1.7.0_79 zookeeper:3.4.9 kafka:2.10-0.10.1.0 一.环境准备[只列,不具 ...
-
CentOS下Hadoop-2.2.0集群安装配置
对于一个刚开始学习Spark的人来说,当然首先需要把环境搭建好,再跑几个例子,目前比较流行的部署是Spark On Yarn,作为新手,我觉得有必要走一遍Hadoop的集群安装配置,而不仅仅停留在本地 ...
-
Hadoop多节点集群安装配置
目录: 1.集群部署介绍 1.1 Hadoop简介 1.2 环境说明 1.3 环境配置 1.4 所需软件 2.SSH无密码验证配置 2.1 SSH基本原理和用法 2.2 配置Master无密码登录所有 ...
-
codis集群安装
在网上找了很多codis的集群安装方法,看起来都是大同小异,本人结合了大多种方法完成了一套自己使用的codis的集群安装,可以供大家学习使用,如果有什么问题或者不懂的地方欢迎指正 1.集群规划: 三台 ...
-
[bigdata] spark集群安装及测试
在spark安装之前,应该已经安装了hadoop原生版或者cdh,因为spark基本要基于hdfs来进行计算. 1. 下载 spark: http://mirrors.cnnic.cn/apache ...
-
(原) 1.2 Zookeeper伪集群安装
本文为原创文章,转载请注明出处,谢谢 Zookeeper伪集群安装 zookeeper单机安装配置可以查看 1.1 zookeeper单机安装 1.复制三份zookeeper,分别为zookeeper ...
随机推荐
-
android 手把手教您自定义ViewGroup(一)
1.概述 在写代码之前,我必须得问几个问题: 1.ViewGroup的职责是啥? ViewGroup相当于一个放置View的容器,并且我们在写布局xml的时候,会告诉容器(凡是以layout为开头的属 ...
-
OracleDBA之表管理
下面是Oracle表管理的部分,用到的测试表是oracle数据库中scott用户下的表做的测试,有的实验也用到了hr用户的数据,以下这些东西是我的麦库上存的当时学Oracle的学习笔记今天拿出来和大家 ...
-
tyvj1102 单词的划分
描述 有一个很长的由小写字母组成字符串.为了便于对这个字符串进行分析,需要将它划分成若干个部分,每个部分称为一个单词.出于减少分析量的目的,我们希望划分出的单词数越少越好.你就是来完成这一划分工作的. ...
-
C# 小规模查找集合性能测试
项目中包含浮点运算,大概每秒 20 - 100 万左右. 其计算结果每秒只包含1000个左右. 因此大量运算是重复性的.程序运行时,cpu 在 3% - 10% 浮动.打算将结果缓存.根据键值索值. ...
-
临时改GCC编译器,重启后失效
临时改GCC编译器,重启后失效.例如,用如下命令: export CROSS_COMPILE= <gcc 文件所在的目录>/arm-linux-gnueabihf- 本例中使用的命令如下: ...
-
Rolls.java (page44)
对象的数组:模拟T次投掷骰子的Counter对象的用例: Rolls.java 如下:所用类列表:Counter, StdOut, StdRandom 见page42 public class Rol ...
-
科讯CMS V9标签清单
全新整理V9标签清单 ====================网站通用标签============== {$GetSiteTitle} 显示网站标题 {$GetSiteName} 显示网站名称 {$G ...
-
android 通过shape设置圆形按钮
Android中常常使用shape来定义控件的一些显示属性来美化UI: shape的常用属性有: (1)solid:填充,设置填充的颜色: (2)stroke:描边,设置边界的宽度.颜色等: (3)c ...
-
qscoj 128 喵哈哈村的魔法源泉(2)(模仿快速幂,好题)
喵哈哈村的魔法源泉(2) 发布时间: 2017年5月9日 20:59 最后更新: 2017年5月9日 21:00 时间限制: 1000ms 内存限制: 128M 描述 喵哈哈村有一个魔法源 ...
-
[Swift]LeetCode754. 到达终点数字 | Reach a Number
You are standing at position 0 on an infinite number line. There is a goal at position target. On ea ...