kafka安装配置

kafka介绍

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。
Apache Kafka与传统消息系统相比，有以下不同：
它被设计为一个分布式系统，易于向外扩展；
它同时为发布和订阅提供高吞吐量；
它支持多订阅者，当失败时能自动平衡消费者；
它将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序

flume

Flume是一个分布式、可靠、和高可用的海量日志聚合的系统，支持在系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。
设计目标：
(1) 可靠性
当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障，从强到弱依次分别为：end-to-end（收到数据agent首先将event写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送。），Store on failure（这也是scribe采用的策略，当数据接收方crash时，将数据写到本地，待恢复后，继续发送），Best effort（数据发送到接收方后，不会进行确认）。
(2) 可扩展性
Flume采用了三层架构，分别为agent，collector和storage，每一层均可以水平扩展。其中，所有agent和collector由master统一管理，这使得系统容易监控和维护，且master允许有多个（使用ZooKeeper进行管理和负载均衡），这就避免了单点故障问题。
(3) 可管理性
所有agent和colletor由master统一管理，这使得系统便于维护。多master情况，Flume利用ZooKeeper和gossip，保证动态配置数据的一致性。用户可以在master上查看各个数据源或者数据流执行情况，且可以对各个数据源配置和动态加载。Flume提供了web 和shell script command两种形式对数据流进行管理。
(4) 功能可扩展性
用户可以根据需要添加自己的agent，collector或者storage。此外，Flume自带了很多组件，包括各种agent（file， syslog等），collector和storage（file，HDFS等）。

kafka安装(单机版)

1. 所需工具

   • JDK

   • kafka安装包(自带zookeeper)

   • zookeeper(可选，集群需要)

   • 操作系统:windows or linux or unix

2. 安装步骤

   1. 安装JDK，自行百度安装

   2. 下载kafka安装包：官网下载，或者用命令行下载

      wget http://www-us.apache.org/dist/kafka/1.1.0/kafka_2.12-1.1.0.tgz

   3. 解压并移动到opt目录

      tar -zxvf kafka_2.12-1.1.0.tgz

      sudo mv kafka_2.12-1.1.0 /opt

   4. kafka目录说明

      • /config 所有相关的配置文件
      • /lib Kafka的依赖包
      • /bin Kafka的启动脚本

   5. 编辑config目录下的server.properties文件，加入以下配置并保存

      port=9092

      host.name=localhost

   image image

6. 启动zookeeper和kafka

   • 启动zookeeper：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
   • 启动kafka： bin/kafka-server-start.sh config/server.properties
   • 查看topic列表：bin/kafka-topics.sh --list --zookeeper localhost:2181
   • 创建单分区单副本的话题(topic)话题名称为test：bin/kafka-topics.sh --create -- zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
   • 创建消费者：bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
   • 此时，命令行会进入运行状态，界面就卡住不动了，然后再新建一个终端窗口，进入kafka目录，继续创建一个生产者：bin/kafka-console-producer.sh --broker-list 172.22.2.140:9092 --topic test

kafka运作原理

• Producer: 发送消息者
• Consumer: 消息消费者
• Consumer Group: 消费者团体
• Broker: Kafka处理消息的节点，一个集群可以包含多个Brokers
• Topic : 发送和消费消息主题
• Partition: 储存Topic消息的分区， 每个Topic可以有多个Partitions
• Partition Replica: 每个消息分区的副本, 每个Partition可以有多个副本，一般是2-3个

kafka的物理架构

image

Producer是通过连接Broker来发送消息的
Producer是向某个Topic来传递送消息的
发送到Topic中的消息是可以给不同Broker处理的
Consumer是连接到某个Broker来监听订阅的Topic的
Brokers是通过Zookeeper来进行管理的，并互相通知各自的运行情况
如果Producer或Consumer连接到的Broker没有相关的Topic的，那么消息会自动路由到相关的Broker, 下一次Producer或者Consumer会自动记住相关的Broker

image

在发送消息后，每个消息会依次排列到每个Partition

消息是可以通过配置来决定要在Partition上保留多久

每个消费者可以从不同队列位置来开始消费消息，并且可以重复消费

image

与flume的整合

• 下载flume,或者使用wget下载

  wget http://www-us.apache.org/dist/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz

• 解压后进入conf目录，新建配置文件hw.conf，内容如下

  a1.sources=r1

  a1.channels=c1

  a1.sinks=k1

  a1.sources.r1.type=exec

  #日志文件路径

  a1.sources.r1.command=tail -F /home/linweijian/abc.log

  a1.sources.r1.channels=c1

  a1.channels.c1.type=memory

  a1.channels.c1.capacity=10000

  a1.channels.c1.transactionCapacity=100

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

  #kafka的topic

  a1.sinks.k1.topic=zhuti1

  #kafka的服务器地址

  a1.sinks.k1.brokerList=172.22.2.140:9092

  a1.sinks.k1.requiredAcks=1

  a1.sinks.k1.batchSize=20

  a1.sinks.k1.channel=c1

• 返回flume的根目录，执行命令(执行该命令前需要先启动kafka和zookeeper)

  > /bin/flume-ng agent -n a1 -c conf -f conf/hw.conf -Dflume.root.logger=INFO,console 
  > 
  

控制台显示如下标识启动成功

image

• 开启一个消费者端

  kafka-console-consumer.sh --topic=zhuti1 --zookeeper localhost:2181

• 创建输出脚本test.sh如下:

  image

        #!/bin/bash  
        for((i=0;i<=1000;i++));
        do echo "kafka_test-"+$i>>/home/linweijian/abc.log;
        done

• 执行该脚本，查看kafka消费者端输出

  image