大数据处理工具Kafka、Zk、Spark

2024-04-05 16:50:00

搭建kafka和zk集群环境

安装环境

MAC操作系统
VMware Fusion虚拟机
3个centos7服务器

安装虚拟机飞机票安装虚拟机Centos系统并安装Docker过程记录

安装包下载

https://kafka.apache.org/downloads.html

服务器环境准备

安装文件上传工具

yum install lrzsz

查看服务器ip

ip addr show

centos-1 192.168.84.128 
centos-2 192.168.84.129
centos-3 192.168.84.130

通过ssh工具连接

这个工具很好用

先安装下centos-1服务器环境

上传kafka安装包

mkdir /opt/kafka
通过rz将压缩包上传kafka_2.10-0.8.2.1.tgz
解压tar xvf kafka_2.10-0.8.2.1.tgz

创建zk数据目录 并设定服务器编号
mkdir /opt/zk_data
cd /opt/zk_data
vi myid
该文件内容为1、2、3分别对应centos-1、centos-2、centos-3

配置zk

kafka安装包中内置zk服务

配置zookeeper.properties

vi /opt/kafka/kafka_2.10-0.8.2.1/config/zookeeper.properties

# zk服务器之间的心跳时间间隔 以毫秒为单位
tickTime=2000
# zk 数据保存目录 zk服务器的ID文件也保存到这个目录下
dataDir=/opt/zk_data/
# zk服务器监听这个端口 然后等待客户端连接
clientPort=2181
# zk集群中follower服务器和leader服务器之间建立
# 初始连接时所能容忍的心跳次数的极限值
initLimit=5
# zk集群中follower服务器和leader服务器之间请求和应答过程中所能容忍的心跳次数的极限值
syncLimit=2
# server.N N代表zk集群服务器的编号
# 服务器IP地址:该服务器于leader服务器的数据交换端口:选举leader服务器时用到的通信端口
server.1=192.168.84.128:2888:3888
server.2=192.168.84.129:2888:3888
server.3=192.168.84.130:2888:3888

配置kafka

配置kafka broker

mkdir /opt/kafka/kafka-logs

vi /opt/kafka/kafka_2.10-0.8.2.1/config/server.properties

#kafka broker的唯一标识 集群中不能重复
broker.id=0
# broker监听端口 用于监听producer或者consumer的连接
port=9092
# 当前broker服务器 ip地址或机器名
host.name=192.168.84.128
#broker作为zk的client 可以连接的zk的地址信息
zookeeper.contact=192.168.84.128:2181,192.168.84.129:2181,192.168.84.130:2181
# 日志保存目录
log.dirs=/opt/kafka/kafka-logs

配置broker地址列表

vi /opt/kafka/kafka_2.10-0.8.2.1/config/producer.properties

# 集群中的broker地址列表
broker.list=192.168.84.128:9092,192.168.84.128:9092,192.168.84.128:9092
# Producer类型 async 异步生产者 sync 同步生产者
producer.type=async

配置consumer

vi /opt/kafka/kafka_2.10-0.8.2.1/config/consumer.properties

# consumer可以连接的zk服务器地址列表
zookeeper.contact=192.168.84.128:2181,192.168.84.128:2181,192.168.84.128:2181

打包配置好的kafka安装包并上传到其他服务器



tar cvf kafka_2.10-0.8.2.1.tar ./kafka_2.10-0.8.2.1

得到kafka_2.10-0.8.2.1.tar

scp ./kafka_2.10-0.8.2.1.tar root@192.168.84.129:/opt/kafka
scp ./kafka_2.10-0.8.2.1.tar root@192.168.84.130:/opt/kafka

传到centos-2和centos-3之后

分别操作

解压
tar xvf kafka_2.10-0.8.2.1.tar 

vi /opt/kafka/kafka_2.10-0.8.2.1/config/server.properties 

文件中的 broker.id 和 host.name

broker.id，可以分别复制 1 和 2
host.name 需要改成当前机器的 IP

安装jdk1.8

每个服务器都需要安装java环境

切换阿里云源

mv /etc/yum.repos.d/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo.backup

wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-7.repo

或者

curl -o /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-7.repo

清除缓存

yum makecache

备注

OpenJDK Development Environment:开发版本带JDK

不要下载 Open JDK runtime Environment只有JRE

安装jdk1.8

yum -y install java-1.8.0-openjdk-devel.x86_64

安装路径
/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.272.b10-1.el7_9.x86_64

全局环境变量配置

vi /etc/profile

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.272.b10-1.el7_9.x86_64
export PATH=$JAVA_HOME/bin:$PATH 
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

. /etc/profile

关闭防火墙

报错信息：

2020-09-14 03:28:23,562 [myid:0] - WARN  [WorkerSender[myid=0]:QuorumCnxManager@588] - Cannot open channel to 3 at election address h6/192.168.1.16:3888
java.net.ConnectException: 拒绝连接 (Connection refused)

此时需要关闭防火墙

sudo systemctl stop firewalld #临时关闭
sudo systemctl disable firewalld #然后reboot 永久关闭
sudo systemctl status  firewalld #查看防火墙状态

分别启动zk

mkdir /opt/kafka/run
mkdir /opt/kafka/run/kafka
mkdir /opt/kafka/run/zk
cd /opt/kafka/run/zk

nohup /opt/kafka/kafka_2.10-0.8.2.1/bin/zookeeper-server-start.sh /opt/kafka/kafka_2.10-0.8.2.1/config/zookeeper.properties &

分别启动kafka

cd /opt/kafka/run/kafka

nohup /opt/kafka/kafka_2.10-0.8.2.1/bin/kafka-server-start.sh /opt/kafka/kafka_2.10-0.8.2.1/config/server.properties &

查看kafka和zk进程是否启动

ps -ef | grep kafka

验证kafka、zk环境是否可用

创建消息主题

/opt/kafka/kafka_2.10-0.8.2.1/bin/kafka-topics.sh --create \
--replication-factor 3 \
--partition 3 \
--topic user-behavior-topic \
--zookeeper 192.168.84.128:2181,192.168.84.129:2181,192.168.84.130:2181

通过console producer生产消息

启动console producer

/opt/kafka/kafka_2.10-0.8.2.1/bin/kafka-console-producer.sh --broker-list 192.168.84.128:9092 --topic user-behavior-topic

通过console consumer消费消息

在另一台机器打开consumer

/opt/kafka/kafka_2.10-0.8.2.1/bin/kafka-console-consumer.sh --zookeeper 192.168.84.129:2181 --topic user-behavior-topic --from-beginning

如果在producer console输入一条消息 能从consumer console看到这条消息就代表安装是成功的

centos-1 生产消息

centos-2 消费消息

说明kafka和zk集群环境是可用的

Spark

安装包下载

https://www.apache.org/dyn/closer.lua/spark/spark-3.0.1/spark-3.0.1-bin-hadoop2.7.tgz

tar xvf spark-3.0.1-bin-hadoop2.7.tgz

依赖条件

jdk1.8
python 2.7.5

部署方式

1、底层资源调度可以依赖外部的资源调度框架:相对稳定的Mesos、Hadoop YARN模式

2、使用spark内建的standalone模式

Local[N]本地模式使用N个线程
Local Cluster[Worker,core,Memory]

伪分布式 可以配置所需要启动的虚拟工作节点的数量

以及每个工作节点所管理的CPU数量和内存尺寸

Spark://hostname:port:Standalone模式

Yarn client

主程序运行在本地具体任务运行在yarn集群

YARN standalone/Yarn cluster

主程序逻辑和任务都运行在YARN集群中

需要部署Spark到相关节点

URL为Spark Master主机地址和端口

Mesos://hostname:port:Mesos 模式

需要部署Spark和Mesos到相关节点

URL为Mesos主机地址和端口

上面的部署方式:实际应用中spark应用程序的运行模式取决于传递给 sparkcontext的master环境变量的值

个别模式还需要依赖辅助程序接口来配合使用

示例代码

examples/src/main

运行脚本

bin/run-example[params]

计算PI

spark-3.0.1-bin-hadoop2.7/bin/run-example SparkPi 10 > Sparkpilog.txt

日志包含两部分

一部分是通用日志信息由一系列脚本及程序产生（计算机信息、spark信息）

另一部分是运行程序的输出结果

计算词数

假设有一个数据文件wordcountdata.txt
统计该文件单词出现的个数
spark-3.0.1-bin-hadoop2.7/bin/run-example JavaWordCount ./wordcountdata.txt

RDD

一个spark的任务对应一个RDD

RDD是弹性分布式数据集即一个RDD代表一个被分区的只读数据集

一个RDD生成

可以来自于内存集合和外部系统
也可通过转换操作来自于其他RDD map filter join

脚本的调用过程

Run-example.sh->load-spark-env.sh->lib 目录下的 jar 包文件->spark-submit.sh->spark-class

Scala

1、
scala最终启动的是jvm线程
所以它可以访问java的库文件 例如java.io.File

2、
通过Main函数的方式启动了一个JVM进程
随后针对该进程又托管了一系列线程级别的操作

3、
scala 简单 轻巧 相对java 非常适合并行计算框架的编写

运行过程

函数

根据现有数据集返回一个新的分布式数据集

由于每个原元素经过func函数转换后组成

flatMap

每一个输入函数会被影射为0到多个输出函数

返回值是一个Seq 而不是单一元素

reduceByKey

在一个(K,V)对的数据集上使用返回一个(K,V)对的数据集

Key相同都会被指定的reduce聚合在一起

总体工作流程

无论本地模式
还是分布式模式

内部程序逻辑结构都是类似的
只是其中部分模块有所简化

本地模式中 集群管理模块被简化为进程内部的线程池

spark环境部署

使用docker部署 spark集群

飞机票安装SBT环境运行Scala项目

结尾

下篇文章通过一个实际案例来介绍下如何使用 spark streaming

案例描述：

假设某论坛需要根据用户对站内网页的
点击量，停留时间，以及是否点赞，
来近实时的计算网页热度，
进而动态的更新网站的今日热点模块，
把最热话题的链接显示其中。

码农公寓

搭建kafka和zk集群环境

安装环境

安装包下载

服务器环境准备

先安装下centos-1服务器环境

配置zk

配置kafka

打包配置好的kafka安装包并上传到其他服务器

安装jdk1.8

关闭防火墙

分别启动zk

分别启动kafka

查看kafka和zk进程是否启动

验证kafka、zk环境是否可用

创建消息主题

通过console producer生产消息

通过console consumer消费消息

Spark

安装包下载

依赖条件

部署方式

示例代码

运行脚本

计算PI

计算词数

RDD

脚本的调用过程

Scala

运行过程

函数

总体工作流程

spark环境部署

结尾

相关文章