flink hadoop zookeeper hbase kafka集群的搭建与部署

vi /etc/sysconfig/network-scripts/ifcfg-ens33 

TYPE="Ethernet"
PROXY_METHOD="none"
BROWSER_ONLY="no"
BOOTPROTO="static"
DEFROUTE="yes"
IPV4_FAILURE_FATAL="no"
IPV6INIT="yes"
IPV6_AUTOCONF="yes"
IPV6_DEFROUTE="yes"
IPV6_FAILURE_FATAL="no"
IPV6_ADDR_GEN_MODE="stable-privacy"
NAME="ens33"
UUID="c35b7341-8921-48f5-ad7a-08cb5af4ba54"
DEVICE="ens33"
ONBOOT="yes"
IPADDR=192.168.2.10
NETMASK=255.255.255.0
GATEWAY=192.168.2.1
DNS1=114.114.114.114
DNS2=8.8.8.8

service network restart

其他3台机器依次将设置为2.11，2.12，2.13。

关闭防火墙

systemctl stop firewalld
systemctl disable firewalld

ssh通信

// 生成密钥
ssh-keygen -t rsa
// 将公钥追加到验证表中
cat id_rsa.pub >> ~/.ssh/authorized_keys 
// 将公钥追加到其他主机验证表中
ssh-copy-id -i ~/.ssh/id_rsa.pub root@slave1
ssh-copy-id -i ~/.ssh/id_rsa.pub root@slave2
ssh-copy-id -i ~/.ssh/id_rsa.pub root@slave3

配置hosts，以便DNS解析主机名

vim /etc/hosts

127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.2.10 master
192.168.2.11 slave1
192.168.2.12 slave2
192.168.2.13 slave3

拷贝给其他主机:
scp /etc/hosts root@slave1:/etc/

1.2 Java环境

1.2.1 安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/java目录下,并解压：

mv jdk-8u202-linux-x64.tar.gz /usr/software/java
tar -zxvf jdk-8u202-linux-x64.tar.gz

1.2.2 设置环境变量

vi /etc/profile

export JAVA_HOME=/usr/software/jdk
export JRE_HOME=$JAVA_HOME/jre
export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
export CLASS_PATH=.:$JAVA_HOME/lib:$JRE_HOME/lib

Esc[:wq]保存后，执行以下命令让其当即生效：

source /etc/profile

输入：

java -version

出现以下信息则表明jdk安装成功：
java version "1.8.0_202"
Java(TM) SE Runtime Environment (build 1.8.0_202-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.202-b12, mixed mode)

2 Hadoop集群

2.1 安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/hadoop目录下,并解压：

mv hadoop-3.0.3.tar.gz /usr/software/hadoop
tar -zxvf hadoop-3.0.3.tar.gz

2.2 设置环境变量

vi /etc/profile

export HADOOP_INSTALL=/usr/software/hadoop
export PATH=$PATH:$HADOOP_INSTALL/bin:$HADOOP_INSTALL/sbin

Esc[:wq]保存后，执行以下命令让其当即生效：

source /etc/profile

2.3 修改启动文件

主要为hadoop指定java环境：

vi /usr/software/hadoop/etc/hadoop/hadoop-env.sh

添加如下内容后保存：
JAVA_HOME=/usr/software/jdk

使其当即生效：
source /usr/software/hadoop/etc/hadoop/hadoop-env.sh

输入：

hadoop version

出现以下信息则表明hadoop安装成功：
Hadoop 2.8.5
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r 0b8464d75227fcee2c6e7f2410377b3d53d3d5f8
Compiled by jdu on 2018-09-10T03:32Z
Compiled with protoc 2.5.0
From source with checksum 9942ca5c745417c14e318835f420733
This command was run using /usr/software/hadoop/share/hadoop/common/hadoop-common-2.8.5.jar

2.4 修改配置文件

core-site.xml

主要配置HDFS的地址和端口号。

vi /usr/software/hadoop/etc/hadoop/core-site.xml

<configuration>
   <!-- 指定HDFS老大(namenode)的通信地址 -->
   <property>
        <name>fs.defaultFS</name>
        <value>hdfs://master:9000</value>
   </property>
   <!-- 指定hadoop运行时产生文件的存储路径 -->
   <property>
        <name>hadoop.tmp.dir</name>
        <value>/usr/software/hadoop/tmp</value>
   </property>
</configuration>

hdfs-site.xml

主要配置分布式文件系统。

vi /usr/software/hadoop/etc/hadoop/hdfs-site.xml

<configuration>
   <!-- 设置namenode的http通讯地址 -->
   <property>
       <name>dfs.namenode.http-address</name>
       <value>master:50070</value>
   </property>
   <!-- 设置secondary namenode的http通讯地址 -->
   <property>
       <name>dfs.namenode.secondary.http-address</name>
       <value>slave1:50090</value>
   </property>
   <!-- 设置namenode的存放路径 -->
   <property>
       <name>dfs.namenode.name.dir</name>
       <value>/usr/software/hadoop/dfs/name</value>
   </property>
   <!-- 设置datanode的存放路径 -->
   <property>
       <name>dfs.datanode.data.dir</name>
       <value>/usr/software/hadoop/dfs/data</value>
   </property>
   <!-- 设置hdfs副本数量 -->
   <property>
       <name>dfs.replication</name>
       <value>3</value>
   </property>
</configuration>

mapred-site.xml

主要是配置JobTracker的地址和端口。

vi /usr/software/hadoop/etc/hadoop/mapred-site.xml

<configuration>
   <!-- 设置框架MR使用YARN -->
   <property>
       <name>mapreduce.framework.name</name>
       <value>yarn</value>
   </property>
</configuration>

yarn-site.xml

主要设置resourcemanager以及reducer取数据的方式。

vi /usr/software/hadoop/etc/hadoop/yarn-site.xml

<configuration>
   <!-- 设置resourcemanager在哪个节点 -->
   <property>
       <name>yarn.resourcemanager.hostname</name>
       <value>master</value>
   </property>
   <!-- 设置reducer取数据的方式是mapreduce_shuffle -->
   <property>
       <name>yarn.nodemanager.aux-services</name>
       <value>mapreduce_shuffle</value>
   </property>

   <property>
       <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
       <value>org.apache.hadoop.mapred.ShuffleHandler</value>
   </property>

</configuration>

master和slaves

vi /usr/software/hadoop/etc/hadoop/master

master

#######################################################

vi /usr/software/hadoop/etc/hadoop/slaves

slave1
slave2
slave3

# 需要注意的是：hadoop3.0之后，默认配置文件中无slaves，以workers替代，设置方式与slaves等同。

2.5 启动集群

格式化集群的文件系统

hadoop namenode -format

启动hadoop集群

start-all.sh

关闭hadoop集群

stop-all.sh

HDFS的web界面端口：50070

YARN的web界面端口：8088

3 Spark安装

3.1 Scala环境

安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/scala目录下,并解压：

mv scala-2.12.7.tgz /usr/software/scala
tar -zxvf scala-2.12.7.tgz
12

设置环境变量

vim /etc/profile

export SCALA_HOME=/usr/software/scala/scala-2.12.7
export PATH=$PATH:$SCALA_HOME/bin
1234

Esc[:wq]保存后，执行以下命令让其当即生效：

source /etc/profile
1

输入：

scala

出现以下信息则表明hadoop安装成功：

Welcome to Scala 2.12.7 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191).
Type in expressions for evaluation. Or try :help.

scala> 
12345678

3.2 Spark集群

安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/spark目录下,并解压：

mv spark-2.3.2-bin-hadoop2.7.tgz /usr/software/spark
tar -zxvf spark-2.3.2-bin-hadoop2.7.tgz
12

设置环境变量

vim /etc/profile

export SPARK_HOME=/usr/software/spark/spark-2.3.2-bin-hadoop2.7
export PATH=$PATH:$SPARK_HOME/bin
1234

Esc[:wq]保存后，执行以下命令让其当即生效：

source /etc/profile
1

配置spark参数

cd /usr/software/spark/spark-2.3.2-bin-hadoop2.7/conf/
cp spark-env.sh.template spark-env.sh
vim spark-env.sh

添加如下内容：
export JAVA_HOME=/usr/software/java/jdk1.8.0_191
export SCALA_HOME=/usr/software/scala/scala-2.12.7
export SPARK_MASTER_IP=xxx.xxx.xxx.xxx
export SPARK_WORKER_MEMORY=8g
export HADOOP_CONF_DIR=/usr/software/hadoop/hadoop-3.0.3/etc/hadoop

vim slaves

slave1
slave2
slave3
12345678910111213141516

3.3 spark集群启动

cd /usr/software/spark/spark-2.3.2-bin-hadoop2.7/sbin/
./start-all.sh

注意：spark和hadoop的启动脚本名称是相同的，又因为hadoop已经将sbin目录配置进Path环境变量中去了，所以启动spark时，需要进入spark的sbin目录。
1234

web界面端口：8080

4 Zookeeper集群

本教程中，我们使用slave1,slave2,slave3三台机器搭建zookeeper集群。

首先在slave1上进行相关安装，然后将配置好的目录复制到其他机器上(slave2, slave3)即可。

安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/zookeeper目录下,并解压：

mv zookeeper-3.4.10.tar.gz /usr/software/zookeeper
tar -zxvf zookeeper-3.4.10.tar.gz

配置文件修改

cd /usr/software/zookeeper/conf
cp ./zoo_sample.cfg ./zoo.cfg
vim zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/software/zookeeper/data  # 修改存放zookeeper数据的目录
clientPort=2181
# 添加3个节点的信息
server.1=slave1:2888:3888
server.2=slave2:2888:3888
server.3=slave3:2888:3888

配置参数说明

tickTime：zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔,也就是说每个tickTime时间就会发送一个心跳。

initLimit：配置zookeeper接受客户端（这里所说的客户端不是用户连接zookeeper服务器的客户端,而是zookeeper服务器集群中连接到leader的follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。

当已经超过10个心跳的时间（也就是tickTime）长度后 zookeeper 服务器还没有收到客户端的返回信息,那么表明这个客户端连接失败。总的时间长度就是 10*2000=20秒。

syncLimit：标识leader与follower之间发送消息,请求和应答时间长度,最长不能超过多少个tickTime的时间长度,总的时间长度就是5*2000=10秒。

dataDir：zookeeper保存数据的目录,默认情况下zookeeper将写数据的日志文件也保存在这个目录里；

clientPort：客户端连接Zookeeper服务器的端口,Zookeeper会监听这个端口接受客户端的访问请求；

server.A=B:C:D中的A是一个数字,表示这个是第几号服务器,B是这个服务器的IP地址，C第一个端口用来集群成员的信息交换,表示这个服务器与集群中的leader服务器交换信息的端口，D是在leader挂掉时专门用来进行选举leader所用的端口。

创建ServerID标识

除了修改zoo.cfg配置文件外,zookeeper集群模式下还要配置一个myid文件,这个文件需要放在dataDir目录下。

其他3台机器依次设置。

vi /usr/software/zookeeper/data/myid
1

同时在slave2,slave3相同路径下创建myid文件，并分别输入2, 3保存。

集群启动

在每台机器上分别执行以下命令：

cd /usr/software/zookeeper/bin/
./zkServer.sh start

./zkServer.sh start-foreground

可以输入以下命令查看机器zookeeper的状态：

[root@slave1 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/software/zookeeper/bin/../conf/zoo.cfg
Mode: follower

可以看出，当前节点为zookeeper的从节点。

5 Hbase集群

本案例基于4台机器搭建Hbase集群：

安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/hbase目录下,并解压：

mv hbase-1.2.8-bin.tar.gz /usr/software/hbase
tar -zxvf hbase-1.2.8-bin.tar.gz

hbase-site.xml

cd /usr/software/hbase/conf
vim hbase-site.xml

<configuration>
    <property>
        <name>hbase.master</name>
        <value>master:60000</value>
        <description>hbase的主节点与端口号</description>
    </property>
    <property>
        <name>hbase.master.maxclockskew</name>
        <value>180000</value>
        <description>时间同步允许的时间差</description>
    </property>
    <property>
        <name>hbase.rootdir</name>
        <value>hdfs://master:9000/hbase</value>
        <description>hbase共享目录，持久化hbase数据</description>
    </property>
    <property>
        <name>hbase.cluster.distributed</name>
        <value>true</value>
        <description>是否为分布式</description>
    </property>
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>slave1,slave2,slave3</value>
        <description>指定zookeeper</description>
    </property>
    <property>
        <name>dfs.replication</name>
        <value>3</value>
        <description>备份数</description>
    </property>
</configuration>

regionservers

cd /usr/software/hbase/conf
vim regionservers 

slave1
slave2
slave3

将配置好的hbase目录同步到另外3台机器。

启动hbase

cd /usr/software/hbase/bin
./start-hbase.sh

启动后，在master节点jps看到HMaster进程，slave节点多出HRegionServer进程。

6 Kafka集群

本案例基于4台机器搭建Kafka集群：

安装包拷贝、解压

将压缩包拷贝至Linux系统中，移动到/usr/software/kafka目录下,并解压：

mv  kafka_2.12-2.0.1.tgz /usr/software/kafka
tar -zxvf  kafka_2.12-2.0.1.tgz

修改配置文件

cd /usr/software/kafka/kafka_2.12-2.0.1/config
vi server.properties 

broker.id=0
listeners=PLAINTEXT://master:9092
# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3
# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/usr/software/kafka/logs
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
sage.max.byte=5242880 
# 消息保存的最大值5M
default.replication.factor=3  
# kafka保存消息的副本数，如果一个副本失效了，另两个还可以继续提供服务
replica.fetch.max.bytes=5242880  
# 取消息的最大直接数
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=slave1:2181,slave2:2181,slave3:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
############################# Group Coordinator Settings #############################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

# 主要修改broker.id，log.dirs，zookeeper.connect

将配置好的hbase目录同步到另外3台机器，并修改配置文件中的broker.id。

其他3台机器依次设置broker.id。

启动

/usr/software/kafka/bin
./kafka-server-start.sh -daemon ../config/server.properties

kafka manager安装

kafka manager安装时默认的Web端口为9000,与hadoop的RPC端口冲突，故启动时需要指定另外一个端口号，如：

bin/kafka-manager -Dhttp.port=9002

7 Flink

7.1 设置环境变量

vi /etc/profile

#flink
export FLINK_HOME=/usr/software/flink
export PATH=$FLINK_HOME/bin:$PATH

新使之生效 source /etc/profile

7.2 配置 cd /usr/software/flink/conf

7.2.1配置文件说明这里面需要我们配置的有：slaves和flink-conf.yaml文件，这里面masters文件是用来配置HA的，只要我们不配置HA的话，就不需要配置masters文件（flink也是master/slave结构，但是对于此时master的选择是执行启动脚本的机器为master）。但是slave需要我们配置，配置对应的主机名即可（伪分布式和分布式的区别也就是实际上slave节点的个数，以及分布式在多个节点上而已）。接下来需要我们配置的就是flink-conf.yaml，flink和spark还是有区别的，spark配置文件分spark-env.sh和spark-default.conf文件，而flink的配置都在flink-conf.yaml中完成配置。

7.2.2修改flink-conf.yaml配置文件，先配置一个简单版本，standalone的模式

# JobManager runs.
jobmanager.rpc.address: cdh1
# The RPC port where the JobManager is reachable.
jobmanager.rpc.port: 6123
# The heap size for the JobManager JVM
jobmanager.heap.size: 1024m
# The heap size for the TaskManager JVM
taskmanager.heap.size: 1024m
# The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.
taskmanager.numberOfTaskSlots: 1
# The parallelism used for programs that did not specify and other parallelism.
parallelism.default: 1
#配置是否在Flink集群启动时候给TaskManager分配内存，默认不进行预分配，这样在我们不适用flink集群时候不会占用集群资源
taskmanager.memory.preallocate: false
# 用于未指定的程序的并行性和其他并行性，默认并行度 
parallelism.default: 2 
 #指定JobManger的可视化端口，尽量配置一个不容易冲突的端口
 jobmanager.web.port: 5566
 #配置checkpoint目录
state.backend.fs.checkpointdir: hdfs://cdh1:9000/flink-checkpoints 
#配置hadoop的配置文件
fs.hdfs.hadoopconf: /usr/local/hadoop/etc/hadoop/ 
#访问hdfs系统使用的
fs.hdfs.hdfssite: /usr/local/hadoop/etc/hadoop/hdfs-site.xml

7.2.3修改slaves和masters2个文件，用来配置taskManager和JobManager信息

[hadoop@cdh1 conf]$ cat slaves 
cdh2
cdh3
cdh4
cdh5
[hadoop@cdh1 conf]$ cat masters 
cdh1:8081

7.2.4配置内容注意 flink-conf.yaml中配置key/value时候在“:”后面需要有一个空格，否则配置不会生效。 #sloves文件配置，填写从节点的ip地址即可

同步信息将flink安装所有信息已经环境信息同步到其他机器上面，这里有几台机器就要执行几次

scp /etc/profile root@cdh3:etc/profile scp -r ./flink-1.7.1 root@cdh3:/usr/local

source /etc/proflie

腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git