本帖最后由 那人好像一条狗 于 2018-6-8 11:52 编辑Kafka-Manager是由Yahoo公司开源的一款基于Web的Kafka管理工具，可以对Kafka集群中的Broker、Topic、Partition、Consumer等进行监控查看和管理。 1 获取Kafka-Manager编译包 Kafka-Manager的官网上只提供了源码包，并没有提供编译包，因此获取Kafka-Manager编译包有两种途径： （1）直接从网上获取其他人自己编译好的Kafka-Manager编译包； （2）从官网下载Kafka-Manager源码包，然后自己手动编译。 此种方式见本文最后，非开发人员可以不关注。 1.1 直接获取已有Kafka-Manager编译包 由于Kafka-Manager是利用SBT进行编译构建的，这个过程由于网络等限制比较麻烦和耗时。可以直接从网上获取别人已经编译好的Kafka-Manager包，这样就直接省去了1.2整个章节中的工作。 Kafka-Manager编译包可以直接从下面这个链接获取： http://blog.wolfogre.com/posts/kafka-manager-download/ 下载完成后，直接到本文中的《2 Kafka-Manager部署运行部分》 2 Kafka-Manager部署运行 由于当前MRS集群中根据是否开启了Kerberos认证分成非安全集群和安全集群两种模式。2.1 非安全模式Kafka集群（未开启Kerberos认证）2.1.1 修改Kafka JMX监控的相关配置 由于当前集群中在安装配置Kafka的监控时，默认的只允许Kafka服务所在的本节点访问当前节点的JMX服务；这样导致运行在其他节点上的Kafka-Manager无法访问JMX服务，从而无法获取相关的监控信息。所以需要进行以下配置修改： 到Manager的管理界面上，修改Kafka服务的服务配置，如下图所示： 将KAFKA_JMX_IP配置项修改为${BROKER_IP} 然后再保存配置，重启Kafka服务，以便让修改的配置生效。 2.1.2 修改Kafka-Manager中关于访问JMX服务的配置 说明： MRS 1.7.1以前版本需要做2.1.2步骤，1.7.1不需要 由于当前集群中的Kafka服务在开启JMX服务时，JMX服务的访问地址的后缀为kafka，Kafka-Manager中的访问JMX服务的地址的后缀为jmxrmi，因此需要修改Kafka-Manager中的访问地址的后缀。 修改kafka-manager-master/app/kafka/manager/jmx/KafkaJMX.scala中的doWithConnection方法中的urlString，即第36行【限Kafka-Manager 2.11-1.3.3.17版本】。 原UrlString：[code]val urlString = s"service:jmx:rmi:///jndi/rmi://$jmxHost:$jmxPort/jmxrmi"[/code] 修改为： [code]val urlString = s"service:jmx:rmi:///jndi/rmi://$jmxHost:$jmxPort/kafka"[/code] 2.1.3 将前面编译生成的kafka-manager-1.3.3.17.zip包进行解压，比如存放在/opt目录下2.1.4 修改Kafka-Manager的配置1、修改Kafka-Manager中关于Kafka集群对应的Zookeeper集群的配置以及与consumer相关的配置：[code]cd /opt/kafka-manager-1.3.3.17 vi conf/application.conf kafka-manager.zkhosts="IP1:24002" vi conf/application.conf kafka-manager.consumer.properties.file=/opt/kafka-manager-1.3.3.17/conf/consumer.properties #该consumer.properties在kafka-manager-1.3.3.17/conf目录中，最好写全路径[/code]2、修改/opt/kafka-manager-1.3.3.17/conf/consumer.properties文件：[code]cd /opt/kafka-manager-1.3.3.17 vi /conf/consumer.properties [/code]3、修改Kafka-Manager相关日志文件的存放路径[code]cd /opt/kafka-manager-1.3.3.17[/code] 修改conf/logback.xml和conf/logger.xml文件 将和两个地方修改为你期望的日志存放路径。 2.1.5 启动Kafka-Manager[code]cd /opt/kafka-manager-1.3.3.17 nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9000 &[/code]如果需要使用JMX Polling，在启动之前先执行以下命令 export JMX_PORT=21006 在浏览器中输入IP1:9000，若出现以下界面则说明Kafka-Manager启动成功。 2.2 安全模式Kafka集群（开启Kerberos认证） 2.2.1 修改Kafka JMX监控的相关配置 由于当前集群中在安装配置Kafka的监控时，默认的只允许Kafka服务所在的本节点访问当前节点的JMX服务；这样导致运行在其他节点上的Kafka-Manager无法访问JMX服务，从而无法获取相关的监控信息。1、到Manager的管理界面上，修改Kafka服务的服务配置，如下图所示： 将KAFKA_JMX_IP配置项修改为${BROKER_IP} 然后再保存配置，重启Kafka服务，以便让修改的配置生效。 2.2.2 修改Kafka-Manager中关于访问JMX服务的配置 说明： MRS 1.7.1以前版本需要做2.2.2步骤，1.7.1不需要 由于当前集群中的Kafka服务在开启JMX服务时，JMX服务的访问地址的后缀为kafka，Kafka-Manager中的访问JMX服务的地址的后缀为jmxrmi，因此需要修改Kafka-Manager中的访问地址的后缀。 修改kafka-manager-master/app/kafka/manager/jmx/KafkaJMX.scala中的doWithConnection方法中的urlString，即第36行【限Kafka-Manager 2.11-1.3.3.17版本】。 原UrlString：[code]val urlString = s"service:jmx:rmi:///jndi/rmi://$jmxHost:$jmxPort/jmxrmi"[/code] 修改为：[code]val urlString = s"service:jmx:rmi:///jndi/rmi://$jmxHost:$jmxPort/kafka"[/code] 2.2.7 启动Kafka-Manager 在启动kafka manager之前先做以下几件事情 1. 修改kafka服务端，将allow.everyone.if.no.acl.found配置成true。原因是Kafka Manager会调用SimpleConsumer API去获取每个consumer的offset，新SimpleConsumer的API不支持认证，即需要开启不认证也能消费。 2. 导出环境变量 export JAVA_OPTS="-Djava.security.auth.login.config=/opt/kafka-manager-1.3.3.17/conf/jaas.conf -Dzookeeper.server.principal=zookeeper/hadoop.hadoop.com -Djava.security.krb5.conf=/opt/huawei/Bigdata/MRS_1.7.0/etc/1_4_KerberosClient/kdc.conf -Dzookeeper.request.timeout=120000 -Dkerberos.domain.name=hadoop.hadoop.com" 注意将路径和kerberos.domain.name替换成实际值 [code]/opt/kafka-manager-1.3.3.17 nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9000[/code] 在浏览器中输入IP1:9000，若出现以下界面则说明Kafka-Manager启动成功。 附录： 利用源码手动编译 Kafka-Manager的编译打包是基于SBT的，因此在编译Kafka-Manager要先预先准备以及Java JDK与SBT。1.2.1 配置Java JDK 这里要求JDK的版本为JDK 1.8。具体的JDK的安装与配置，比较常见，这里就不赘述。1.2.2 安装SBT 到SBT官网：http://www.scala-sbt.org/download.html 下载SBT压缩包。将下载的SBT压缩包进行解压并配置SBT环境变量。【目前先讨论下Windows环境下的编译】。 本文主要是针对Windows操作系统下的SBT的安装，至于Linux操作系统下SBT的安装可自行搜索安装。1. Windows环境： 假设解压的SBT放在D:\sbt目录。 添加SBT_HOME环境变量，[code]SBT_HOME ="D:\sbt"[/code] 并在Path中追加[code];%SBT_HOME%\bin;[/code] 然后通过在命令行窗口中输入sbt，如果sbt被识别则说明已经安装配置成功。1.2.3 配置SBT 由于在利用SBT工具编译项目时，需要下载很多SBT工程编译时依赖的jar包等资源，由于SBT默认的资源都是在国外，由于网络原因可以会很慢甚至无法连接，所以为了能更快速地下载这些资源，可以对SBT进行一些配置。1、 先配置SBT的本地资源存放目录，以便在下载一次后可以为之后的编译使用。 打开“SBT安装目录:\conf\sbtconfig.txt”文件，在文件最后追加[code]-Dsbt.boot.directory=D:/boot/ -Dsbt.global.base=D:/.sbt -Dsbt.ivy.home=D:/.ivy2[/code]2、 在“SBT安装目录:\conf\”目录下，新建文件repo.properties文件，将以下内容复制到文件中。[code][repositories] local store_1:http://repo.typesafe.com/typesafe/ivy-releases/ store_2:http://repo2.maven.org/maven2/ typesafe: http://repo.typesafe.com/typesafe/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext], bootOnly typesafe2: http://repo.typesafe.com/typesafe/releases/ sbt-plugin: http://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/ sonatype: http://oss.sonatype.org/content/repositories/snapshots uk_maven: http://uk.maven.org/maven2/ ibibli: http://mirrors.ibiblio.org/maven2/ repo2: http://repo2.maven.org/maven2/ aliyun: http://maven.aliyun.com/nexus/content/groups/public comp-maven:http://mvnrepository.com/artifact/ store_cn:http://maven.oschina.net/content/groups/public/ store_mir:http://mirrors.ibiblio.org/maven2/ store_0:http://maven.net.cn/content/groups/public/ sbt-releases-repo: http://repo.typesafe.com/typesafe/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext] sbt-plugins-repo: http://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext] maven-central: http://repo1.maven.org/maven2/[/code]3、 打开“SBT安装目录:\conf\sbtconfig.txt”文件，在文件最后追加：[code]-Dsbt.repository.config= SBT安装目录:/conf/repo.properties[/code]4、 如果依旧有一些repo无法访问，可以考虑为SBT设置代理，由于走HTTPS时，由于证书问题，还是走HTTP较好。 打开“SBT安装目录:\conf\sbtconfig.txt”文件，在文件最后追加[code]-Dsbt.repository.secure=false #配置项告知SBT使用HTTP而非HTTPS来更新和下载artifacts -Dhttp.proxyHost= proxy host -Dhttp.proxyPort= proxy port -Dhttp.proxyUser=用户名 -Dhttp.proxyPassword=密码[/code] 1.2.4 利用SBT编译Kafka-Manager1、下载Kafka-Manager工程 到https://github.com/yahoo/kafka-manager界面上直接下载Kafka-Manager工程。2、将步骤1中下载的kafka-manager-master.zip文件进行解压，并进入kafka-manager-master目录，如：D:\kafka-manager-master。[code]cd D:\kafka-manager-master sbt clean dist[/code] 该步骤由于需要下载很多的jar包等资源文件，比较耗时，需要耐心等待。 如果最后出现如下输出，则说明Kafka-manager编译成功。 3、到D:\kafka-manager-master\target\universal目录中找到编译好的kafka-manager包即kafka-manager-1.3.3.17.zip。

摘要: 实时数据流是现在互联网公司、甚至拥有大规模数据的传统企业的主要模式， 实时数据(Real-time Activity Data)就是那些非交易，不需要秒级响应的数据， 但在后续的分析中产生极大作用，例如广播、排序、个性化推荐、运营监控、报表等 Linkin的Kafka主要就是为了处理这种性质的数据设计的。 本文是对Linkin的论文的学习笔记，欢迎大家吐槽！ Kafka是啥？##是个消息中间件吗？那和市面上其他一堆堆的中间件例如ActiveMQ, RabbitMQ有什么区别？答案只有一个：Kafka是个集群的消息中间件+存储，一个节点可以存储几T的数据！为啥一个中间件需要存储数据呢？慢慢道来……原来，对于Linkin这样的互联网企业来说，用户和网站上产生的数据有三种： [*]需要实时响应的交易数据，用户提交一个表单，输入一段内容，这种数据最后是存放在关系数据库(Oracle, MySQL)中的，有些需要事务支持。 [*]活动流数据，准实时的，例如页面访问量、用户行为、搜索情况，这些数据可以产生啥？广播、排序、个性化推荐、运营监控等。这种数据一般是前端服务器先写文件，然后通过批量的方式把文件倒到Hadoop这种大数据分析器里面慢慢整。 [*]各个层面程序产生的日志，例如httpd的日志、tomcat的日志、其他各种程序产生的日志。码农专用，这种数据一个是用来监控报警，还有就是用来做分析。 Linkin的牛逼之处，就在于他们发现了原先2,3的数据处理方式有问题，对于2而言，原来动辄一两个钟头批处理一次的方式已经不行了，用户在一次购买完之后最好马上就能看到相关的推荐。而对于3而言，传统的syslog模式等也不好用，而且很多情况下2和3用的是同一批数据，只是数据消费者不一样。这2种数据的特点是： [*]准实时，不需要秒级响应，分钟级别即可。 [*]数据量巨大，是交易数据的10倍以上。 [*]数据消费者众多，例如评级、投票、排序、个性化推荐、安全、运营监控、程序监控、后期报表等 于是，Linkin就自己开发了一套系统，专门用来处理这种性质的数据，这就是Kafka那么，在整个实践过程中Linkin做了什么样的设计，解决了什么问题？首先看下数据流动图：5817多数据中心怎么管理数据：5818集群本身的架构图5819Kafka内部架构图，分为数据产生者(Producer)，数据中间者(Broker)，数据消费者(Consumer)显然，这是一个集群的发布/订阅系统，有如下几个特点 [*]生产者是推数据(Push)，消费者是拉数据(Pull)。存在数据复用，在Linkin平均生产1条消息会被消费5.5次。 [*]数据生产者和数据消费者的速度不对等，所以要把数据沉淀在Kafka内慢慢处理，Linkin一般在集群内放7天的数据。 [*]性能上追求高吞吐，保证一定的延时性之内。这方面做了大量优化，包括没有全局hash，批量发送，跨数据中心压缩等等。 [*]容错性上使用的“至少传输一次”的语义。不保证强一次，但避免最多传一次的情况。 [*]集群中数据分区，保证单个数据消费者可以读到某话题(topic)的某子话题(例如某用户的数据)的所有数据，避免全局读数据 [*]数据规范性，所有数据分为数百个话题，然后在数据的源头——生产者(Producer)这边就用Schema来规范数据，这种理念使得后期的数据传输、序列化、压缩、消费都有了统一的规范，同时也解决了这个领域非常麻烦的数据版本不兼容问题——生产者一改代码，消费者就抓瞎。 [*]用于监控，这个系统的威力在于，前面所有生产系统的数据流向，通过这个系统都能关联起来，用于日常的运营也好，用于数据审计，用于运维级别的监控也好都是神器啊！ To be continued...##所以，Kafka的设计基本上目前这个领域的唯一选择。我也看了很多其他实现，包括：5820 [*]数据采集组件 [*]数据传输组件 [*]数据实时计算/索引/搜索组件 [*]数据存储/持久化组件 [*]数据展示/查询/报警界面组件 从数据传输这块的设计理念来说，Kafka是最为先进的，在目前的各种实现中，我猜测可以和Kafka一战的也就只有Splunk了后面我会分析一下这个软件的设计和实现欲知后事如何，且听下回分解 ~~主要参考文章日志：每个软件工程师都应该知道的有关实时数据的统一概念 —— 这篇比较抽象，高屋建瓴，理论先行Building LinkedIn’s Real-time Activity Data Pipeline —— 实践层的论文，把做事情的前因后果都写明白了分布式发布订阅消息系统 Kafka 架构设计 —— 落地设计次要参考文章《分布式发布订阅消息系统 Kafka 架构设计》《StreamBase简介》《Yahoo! s4和Twitter storm的粗略比较》《最火爆的开源流式系统Storm vs 新星Samza》《架构之淘宝实时数据传输平台: TimeTunnel介绍》《Graylog2 简介》《logstash 还是不行》《日志收集以及分析：Splunk 》《LogStash日志分析系统》《LogStash，使日志管理更简单》《logstash VS splunk》《个性化离线实时分析系统pora》《日志：每个软件工程师都应该知道的有关实时数据的统一概念》《基于Flume的美团日志收集系统(二)改进和优化》《基于Flume的美团日志收集系统(一)架构和设计》《对互联网海量数据实时计算的理解》《流式日志系统启示录》《flume-ng+Kafka+Storm+HDFS 实时系统搭建》

本帖最后由 yd_55568363 于 2017-11-20 14:41 编辑1 问题描述： 开源confluent-kafka-go连接华为MRS的安全集群失败。 具体原因：confluent-kafka-go依赖的库librdkafka默认将broker所在hostname作为了server princple的一部分来使用，导致认证失败。 2 修改思路： 增加domainName参数用于认证，允许客户端设置。 2.1 librdkafka具体修改步骤： 源码地址：https://github.com/edenhill/librdkafka src/rdkafka_conf.c 增加sasl.kerberos.service.name配置项 "Kerberos principal name that Kafka runs as.", .sdef = "kafka" }, { _RK_GLOBAL, "sasl.kerberos.principal", _RK_C_STR, _RK(sasl.principal), "This client´s Kerberos principal name.", .sdef = "kafkaclient" }, + { _RK_GLOBAL, "sasl.kerberos.domain.name", _RK_C_STR, + _RK(sasl.domain_name), + "This cluster´s Kerberos domain name.", + .sdef = "hadoop.hadoop.com" }, #ifndef _MSC_VER { _RK_GLOBAL, "sasl.kerberos.kinit.cmd", _RK_C_STR, _RK(sasl.kinit_cmd), "Full kerberos kinit command string, %{config.prop.name} is replaced " "by corresponding config object value, %{broker.name} returns the " "broker´s hostname.", - .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{broker.name}\" " + .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{sasl.kerberos.domain.name}\" " "-k -t \"%{sasl.kerberos.keytab}\" %{sasl.kerberos.principal}" }, { _RK_GLOBAL, "sasl.kerberos.keytab", _RK_C_STR, _RK(sasl.keytab), "Path to Kerberos keytab file. Uses system default if not set." "**NOTE**: This is not automatically used but must be added to the " "template in sasl.kerberos.kinit.cmd as " src/rdkafka_conf.h增加domain_name字段 --- src\rdkafka_conf.h 2017-10-17 11:20:56.000000000 +0800 +++ src\rdkafka_conf.h 2017-10-25 16:26:34.000000000 +0800 @@ -118,12 +118,13 @@ struct { const struct rd_kafka_sasl_provider *provider; char *principal; char *mechanisms; char *service_name; + char *domain_name; char *kinit_cmd; char *keytab; int relogin_min_time; char *username; char *password; #if WITH_SASL_SCRAM src/rdkafka_sasl_cyrus.c 将hostname替换成domainName --- src\rdkafka_sasl.c 2017-10-17 11:20:56.000000000 +0800 +++ src\rdkafka_sasl.c 2017-10-25 16:09:38.000000000 +0800 @@ -192,13 +192,14 @@ rk->rk_conf.sasl.mechanisms, rk->rk_conf.api_version_request ? "" : ": try api.version.request=true"); return -1; } - rd_strdupa(&hostname, rktrans->rktrans_rkb->rkb_nodename); + //rd_strdupa(&hostname, rktrans->rktrans_rkb->rkb_nodename); + rd_strdupa(&hostname, rk->rk_conf.sasl.domain_name); if ((t = strchr(hostname, ´:´))) *t = ´\0´; /* remove ":port" */ 3 重新编译librdkafka 具体步骤参考https://github.com/edenhill/librdkafka/tree/v0.11.1 要求：已安装libsasl2-dev 编译： ./configure make make install 4 客户端使用 增加如下配置项： "security.protocol": "SASL_PLAINTEXT", "sasl.kerberos.service.name": "kafka", "sasl.kerberos.keytab": "/opt/nemon/user.keytab", ----需要换成路径 "sasl.kerberos.principal": "nemon@HADOOP.COM",----需要换成实际用户名 "sasl.kerberos.domain.name": "hadoop.hadoop.com",--需要换成实际domainName 其中前四项为开源已有参数。Keytab可通过manager界面下载。系统设置—用户管理—更多—下载认证凭据。 domainName如何获取： domain的命名规则为：hadoop. toLowerCase(realm)。假设集群的域名（default_realm）为HUAWEI.COM时，domain的值为hadoop.huawei.com。 可通过manager界面服务管理—KrbServer—服务配置—参数类别(全部配置)—域 ，查看default_realm。 示例代码consumer_example.go(非安全版VS安全版)： 运行效果：略