目录
1、创建输入主题：streams-plaintext-input
2、创建输出主题：streams-wordcount-output
一、Zookeeper安装
参考上篇写的zookeeper安装文章。
二、kafka安装
1、下载地址
????????安装的 Kafka 版本为 kafka_2.11-2.4.0.tgz，其中 2.11 代表 Scala 版本， 2.4.0 表示 Kafka 的版本。
2、kafka解压安装并移动
tar xvf kafka_2.11-2.4.0.tgz
mv kafka_2.11-2.4.0/ /usr/local/kafka/3、环境变量配置
vim ~/.basbrc
export KAFKA_HOME=/usr/local/kafka/
PATH=\$PATH:\$JAVA_HOME/bin:\$JRE_HOME/bin:\$ZOOKEEPER_HOME/bin:\$KAFKA_HOME/bin4、修改配置文件
????????修改$KAFKA_HOME/config 目录下的 server.properties文件，为了便于后 续集群环境搭建的配置， 需要保证同一个集群下 broker.id要唯一，因此这里手动配置 broker.id, 直接保持与ZooKeeper的myid值一致，同时配置日志存储路径。
server.properties配置如下 :
broker.id=l #指定代理的 id
log.dirs=/usr/local/kafka/kafka-logs #指定 Log 存储路径
zookeeper.connect=master:2181,node1:2181,node2:2181zookeeper.properties配置如下:（dataDir和clientPort要和zookeeper配置的一致）
dataDir=/usr/local/zookeeper/dataconsumer.properties配置如下：
bootstrap.servers=master:9092,node1:9092,node2:9092producer.properties配置如下：
bootstrap.servers=master:9092,node1:9092,node2:9092????????将kafka文件夹scp到其他节点上，在三台机器上分别修改配置文件server.properties, 并修改对应的broker.id。
5、启动
./kafka-server-start.sh -daemon ../config/server.properties执行 jps命令查看 Java进程，此时进程信息至少包括以下几项:
15976 Jps
14999 QuorumPeerMain
15906 Kafka三、kafka和spark集成
1、下载安装包
spark-streaming-kafka-0-10_2.11-2.4.4.jar
在maven仓库下载安装包，下载地址：Maven Central Repository Search
https://search.maven.org/
选择适合的版本
?将下载好的包放在spark安装目录下：/usr/local/spark/jars/
2、kafka-clients包复制
将kafaka安装包/usr/local/kafka/libs/下的kafka-clients-2.4.0.jar复制到/usr/local/spark/jars/一份
cp /usr/local/kafka/libs/kafka-clients-2.4.0.jar /usr/local/spark/jars/四、kafka实例
1、创建输入主题：streams-plaintext-input
bin/kafka-topics.sh --create \
--bootstrap-server master:9092 \
--replication-factor 1 \
--partitions 1 \
--topic streams-plaintext-input2、创建输出主题：streams-wordcount-output
由于输出流是一个 changelog 流，启用压缩。
bin/kafka-topics.sh --create \
--bootstrap-server master:9092 \
--replication-factor 1 \
--partitions 1 \
--topic streams-wordcount-output \
--config cleanup.policy=compact使用 kafka-topics 工具，查看已创建主题的描述：
bin/kafka-topics.sh --bootstrap-server master:9092 --describe3、Wordcount 应用程序测试
1）启动 Wordcount 应用程序
示例程序将从输入主题 streams-plaintext-input 中读取消息，在执行 WordCount 算法计算之后，将结果连续写入输出主题 streams-wordcount-output。
bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
2）启动控制台生产者，向输入主题中写入数据：
bin/kafka-console-producer.sh --broker-list master:9092 --topic streams-plaintext-input
3）使用控制台消费者，从输出主题中读取结果：
bin/kafka-console-consumer.sh --bootstrap-server master:9092 \
--topic streams-wordcount-output \
--from-beginning \
--formatter kafka.tools.DefaultMessageFormatter \
--property print.key=true \
--property print.value=true \
--property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
--property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
结语
?????????本次的分享就到这里了，下一篇博客博主将带来kafka实战，敬请期待！受益的朋友或对大数据技术感兴趣的伙伴记得点赞关注支持一波！