数据实时同步平台搭建-阿里云开发者社区

执行安装命令，安装openjdk8

yum install java-1.8.0-openjdk

查看路径，找到Java的具体安装路径

which java
ls -lrt /usr/bin/java

进入上一步找到的Java目录，查看java是否安装成功

java -version

安装Javac

yum install java-1.8.0-openjdk-devel.x86_64

配置Java环境变量,编辑/etc/profile文件，在末尾追加如下内容

JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-xxxx 这就是刚刚说的那个路径的名称
JRE_HOME=$JAVA_HOME/jre
CLASS_PATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
export JAVA_HOME JRE_HOME CLASS_PATH PATH

二、MySQL部署

在/usr/local目录下创建mysql文件夹。

mkdir mysql

下载mysql8.0安装包,下载地址：https://dev.mysql.com/downloads/mysql/
上传mysql-8.0.30-linux-glibc2.12-x86_64.tar.xz到/usr/local/mysql目录下面。
解压文件并重命名。

tar xJf mysql-8.0.30-linux-glibc2.12-x86_64.tar.xz
mv mysql-8.0.30-linux-glibc2.12-x86_64 mysql-8.0

在/usr/local/mysql目录下创建data文件夹，用于存放mysql数据。

mkdir data

分别创建用户组以及用户和密码

groupadd mysql
useradd -g mysql mysql

授权刚刚新建的用户

chown -R mysql.mysql /usr/local/mysql/mysql-8.0
chmod 750 /usr/local/mysql/data -R

配置环境变量。修改/etc/profile文件，在末尾追加。

export PATH=$PATH:/usr/local/mysql/mysql-8.0/bin:/usr/local/mysql/mysql-8.0/lib

编辑数据库配置文件 /etc/my.cnf

[mysql]
default-character-set=utf8mb4
[client]
socket=/var/lib/mysql/mysql.sock
[mysqld]
user=mysql
general_log = 1
general_log_file= /var/log/mysql/mysql.log
socket=/var/lib/mysql/mysql.sock
basedir=/usr/local/mysql/mysql-8.0
datadir=/usr/local/mysql/data
log-bin=/usr/local/mysql/data/mysql-bin
innodb_data_home_dir=/usr/local/mysql/data
innodb_log_group_home_dir=/usr/local/mysql/data/
character-set-server=utf8mb4
lower_case_table_names=1
autocommit=1
default_authentication_plugin=mysql_native_password
symbolic-links=0
[mysqld_safe]
log-error=/usr/local/mysql/data/mysql.log
pid-file=/usr/local/mysql/data/mysql.pid

初始化MySQL信息,得到数据库的初始密码（在/usr/local/mysql/mysql-8.0/bin目录下执行）

./mysqld --user=mysql --basedir=/usr/local/mysql/mysql-8.0 --datadir=/usr/local/mysql/data/ --initialize

复制 mysql.server 文件，在/usr/local/mysql/mysql-8.0目录下执行。

cp -a ./support-files/mysql.server /etc/init.d/mysql
cp -a ./support-files/mysql.server /etc/init.d/mysqld

赋予权限

chown 777 /etc/my.cnf
chmod +x /etc/init.d/mysql
chmod +x /etc/init.d/mysqld

检查一下/var/lib/mysql是否存在，否则进行创建并授权。

mkdir /var/lib/mysql
chown -R mysql:mysql /var/lib/mysql/

启动数据库。

service mysql start

修改root账号密码。

mysql -h127.0.0.1 -uroot -p
ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY '新密码';
flush privileges;

三、zookeeper部署

zookeeper集群部署

0.下载zookeeper3.8.0，下载地址：https://www.apache.org/dyn/closer.lua/zookeeper/zookeeper-3.8.0/apache-zookeeper-3.8.0-bin.tar.gz

将zookeeper解压到/usr/local目录下面(zookeeper要下载名称中带-bin的包)。

tar -zxf apache-zookeeper-3.8.0-bin.tar.gz

在zookeeper下面创建zkData目录。

mkdir zkData

重命名conf目录下面的配置文件zoo_simple.cfg

mv zoo_simple.cfg zoo.cfg

修改zoo.cfg文件。

具体配置 
dataDir=/usr/local/zookeeper/zkData 
增加如下配置 
#######################cluster########################## 
server.1=test-bigdata1:2888:3888
server.2=test-bigdata2:2888:3888
server.3=test-bigdata3:2888:3888

touch myid
在myid里面写入1,表示zoo.cfg里面server.1中的1。

将配置好的zookeeper拷贝到另外两台服务器，修改myid文件中的值为对象的server的值。
分别启动zookeeper。

bin/zkServer.sh start

查看启动状态，有一台leader，两天follower，表示启动成功。

bin/zkServer.sh status
ZooKeeper JMX enabled by default
Client port found: 2181. Client address: localhost.
Mode: follower
bin/zkServer.sh status
ZooKeeper JMX enabled by default
Client port found: 2181. Client address: localhost.
Mode: leader
bin/zkServer.sh status
ZooKeeper JMX enabled by default
Client port found: 2181. Client address: localhost.
Mode: follower

** 如果启动过程中出现org.apache.zookeeper.server.quorum.QuorumPeerConfig$ConfigException: Address unresolved: bigdata03:3888 异常，是因为bigdata03:3888后面有空格所致。**

部署zkui可视化页面

在GITHUB下载源码打包编译，https://github.com/DeemOpen/zkui

mvn clean package

打包完成会在target目录下面生成zkui-2.0-SNAPSHOT-jar-with-dependencies.jar文件。
将zkui-2.0-SNAPSHOT-jar-with-dependencies.jar和config.cfg文件拷贝到服务器。
创建需要的数据库和用户，并初始化。

mysql -h127.0.0.1 -uroot -p 
# 创建账号并修改密码
CREATE user 'zkui'@'%';
alter user 'zkui'@'%' identified with mysql_native_password by '密码';
#创建数据库
create database zkui;
# 授权数据库
grant all privileges on zkui.* to "zkui"@"%";
# 刷新权限
flush privileges;

修改config.cfg文件，配置相应的数据库信息。

zkServer=test-bigdata1:2181,test-bigdata2:2181,test-bigdata3:2181
jdbcClass=com.mysql.jdbc.Driver
jdbcUrl=jdbc:mysql://test-bigdata4:3306/zkui?characterEncoding=utf8&useSSL=false
jdbcUser=zkui
jdbcPwd=密码

创建启动脚本，新建start.sh内容如下。

nohup java -jar zkui-2.0-SNAPSHOT-jar-with-dependencies.jar >2&1 >logs/zkui-out.log &

创建停止脚本，新建stop.sh内容如下。

#! /bin/bash
PROGRAM_PID=`ps -ef | grep zkui-2.0-SNAPSHOT-jar-with-dependencies.jar| grep -v grep | awk '{print $2}'` 
if [[ ${PROGRAM_PID}x != ""x ]]; then
        echo "stop service at PID[${PROGRAM_PID}] ..."
        kill -9 ${PROGRAM_PID}
        if [[ $? -eq 0 ]]; then
                echo "stop service at PID[${PROGRAM_PID}] success"
        else
                echo "stop service at PID[${PROGRAM_PID}] error"
        fi
else
        echo "service is not running."
fi

启动

./start.sh

查看页面

http://192.168.20.167:9090/home
账号密码:admin/manager

四、Pulsar部署

下载pulsar，下载地址：http://archive.apache.org/dist/pulsar/pulsar-2.10.1/apache-pulsar-2.10.1-bin.tar.gz
将其上传到/usr/local目录下面
解压

tar -zxf apache-pulsar-2.10.1-bin.tar.gz

配置环境变量，修改/etc/profile文件，在其末尾追加。

export PULSAR_HOME=/opt/local/apache-pulsar-2.10.1
export PATH=$PATH:$PULSAR_HOME/bin

初始化集群元数据。

pulsar initialize-cluster-metadata \
  --cluster pulsar-cluster-1 \
  --zookeeper 192.168.20.164:2181 \
  --configuration-store 192.168.20.164:2181 \
  --web-service-url http://192.168.20.164:8070,192.168.20.165:8070,192.168.20.166:8070 \
  --broker-service-url pulsar://192.168.20.164:6650,192.168.20.165:6650,192.168.20.166:6650

参数说明如下：

参数	说明
—cluster	pulsar 集群名字
–zookeeper	zookeeper 地址，只需要包含 zookeeer 集群中的任意一台机器即可
–configuration-store	配置存储地址，只需要包含 zookeeer 集群中的任意一台机器即可
–web-service-url	pulsar 集群 web 服务的 URL 以及端口，默认的端口是8080,自定义为8070
–broker-service-url	broker 服务的URL，用于与 pulsar 集群中的 brokers 进行交互，默认端口是 6650

部署BookKeeper集群

修改BookKeeper配置文件

vim bookkeeper.conf 
zkServers=192.168.20.164:2181,192.168.20.165:2181,192.168.20.166:2181

mkdir -p /usr/local/apache-pulsar-2.10.1/data/bookkeeper

初始化BookKeeper元数据，(只需在一个bookie节点执行一次)

bookkeeper shell metaformat

启动BookKeeper集群。三台分别启动

pulsar-daemon start bookie

验证集群状态，在任意一台 Bookeeper 节点上使用 Bookeeper shell 的 simpletest 命令，去校验集群内所有的 bookie 是否都已经启动，3 为 Bookeeper 节点数量。

bookkeeper shell simpletest --ensemble 3 --writeQuorum 3 --ackQuorum 3 --numEntries 3

参数含义如下： -a,–ackQuorum Ack quorum size (default 2) 当指定数量的 bookie ack 响应时，认为消息写入成功 -e,–ensemble Ensemble size (default 3) 写入数据的 bookie 节点数量 -n,–numEntries Entries to write (default 1000) 一批消息的消息数量 -w,–writeQuorum Write quorum size (default 2) 每条消息副本数量这个命令会在集群上创建和 bookie 同等数量的 ledger，并往里面写一些条目，然后读取它，最后删除这个 ledger。

部署pulsar集群

修改pulsar配置文件。三台均需要

vim broker.conf
# 配置pulsar broker连接的zookeeper集群地址
zookeeperServers=192.168.20.164:2181,192.168.20.165:2181,192.168.20.166:2181
configurationStoreServers=192.168.20.164:2181,192.168.20.165:2181,192.168.20.166:2181
clusterName=pulsar-cluster-1

启动Pulsar集群，三台均需要。

pulsar-daemon start broker

配置客户端连接pulsar集群。

vim client.conf
webServiceUrl=http://192.168.20.164:8070,192.168.20.165:8070,192.168.20.166:8070
brokerServiceUrl=pulsar://192.168.20.164:6650,192.168.20.165:6650,192.168.20.166:6650

命令行验证生产消费消息。

启动消费者

pulsar-client consume \
  persistent://public/default/pulsar-test \
  -n 100 \
  -s "consumer-test" \
  -t "Exclusive"

启动生产者

pulsar-client produce \
  persistent://public/default/pulsar-test \
  -n 1 \
  -m "Hello Pulsar"

观察消费端控制台输出，如打输出内容content:Hello Pulsar，则流程全部完成；

部署PulsarManager

下载pulsar-manager的源码，https://github.com/apache/pulsar-manager。
修改源代码的build.gradle文件，新增MySQL依赖。

compile group: 'mysql', name: 'mysql-connector-java', version: mysqlVersion

修改源码的gradle.properties文件，新增MySQL版本。

mysqlVersion=8.0.25

执行打包命令，打包文件。

gradlew.bat distTar -x test

打包成功后，在build\distributions目录下面会生成pulsar-manager.tar文件。
下载apache-pulsar-manager-0.3.0-bin.tar.gz。下载地址：https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-manager/pulsar-manager-0.3.0/apache-pulsar-manager-0.3.0-bin.tar.gz。
将apache-pulsar-manager-0.3.0-bin.tar.gz和pulsar-manager.tar全都上传到/usr/local目录下面
解压文件。

tar -zxf apache-pulsar-manager-0.3.0-bin.tar.gz -c /usr/local/test
tar -zxf pulsar-manager.tar

将 apache-pulsar-manager-0.3.0-bin.tar.gz解压包里面的dist目录。拷贝到pulsar-manager下面。

cp -r /usr/local/test/pulsar-manager/dist /usr/local/pulsar-manager/ui

在MySQL中创建数据库和账号。

mysql -h127.0.0.1 -uroot -p 
# 创建账号并修改密码
CREATE user 'pulsar'@'%';
alter user 'pulsar'@'%' identified with mysql_native_password by '密码';
#创建数据库
create database pulsar_manager;
# 授权数据库
grant all privileges on pulsar_manager.* to "pulsar"@"%";
# 刷新权限
flush privileges;

修改配置文件application.properties。

# mysql configuration
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://test-bigdata4:3306/pulsar_manager?characterEncoding=utf8&useSSL=false
spring.datasource.username=pulsar
spring.datasource.password=密码

创建一个start.sh并授权，作为启动

nohup bin/pulsar-manager-0.3.0 >> logs/pulsar-manager.log 2>&1 &
chmod +x start.sh

启动

./start.sh

初始化密码.

# 执行获取csrf-token
curl http://192.168.20.167:7750/pulsar-manager/csrf-token
# 使用上一步获取的token替换CSRF_TOKEN值，然后执行
curl -H 'X-XSRF-TOKEN: $CSRF_TOKEN' -H 'Cookie: XSRF-TOKEN=$CSRF_TOKEN;' -H "Content-Type: application/json" -X PUT http://192.168.20.167:7750/pulsar-manager/users/superuser -d '{"name": "pulsar", "password": "pulsar", "description": "superuser", "email": "surperuser@pulsar.org"}'

查看页面

http://192.168.20.167:7750/ui/index.html
账号密码:pulsar/pulsar

新增环境

image.png

五、HADOOP部署

修改配置文件

安装hadoop，选择安装hadoop3.2.1。下载地址：https://archive.apache.org/dist/hadoop/common/hadoop-3.1.0/hadoop-3.1.0.tar.gz
解压文件到/usr/local目录下面

tar -zxf hadoop-3.2.1.tar.gz -C /usr/local

修改hadoop的etc/hadoop目录下面的hadoop-env.sh文件。

# 在其中加入jdk路,替换为自己真实的地址
JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-xxxx
# 加入各个组件操作的用户，防止放弃操作出现用户无法操作问题。
export HDFS_NAMENODE_USER=root
export HDFS_DATANODE_USER=root
export HDFS_SECONDARYNAMENODE_USER=root
export YARN_RESOURCEMANAGER_USER=root
export YARN_NODEMANAGER_USER=root
export HDFS_JOURNALNODE_USER=root
export HDFS_ZKFC_USER=root

修改的etc/hadoop目录下面的core-site.xml文件。

<configuration>
   <!--把两个NameNode的地址组装成一个集群mycluster -->
   <property>
      <name>fs.defaultFS</name>
      <value>hdfs://mycluster</value>
   </property>   
   <!--指定hadoop运行时产生文件的存储目录 -->
   <property>
      <name>hadoop.tmp.dir</name>
      <value>/usr/local/hadoop-3.2.1/data/tmp</value>
   </property>
   <!--配置zookeeper集群的地址,进行namenode发生故障时自动转移 -->
   <property>
      <name>ha.zookeeper.quorum</name>
      <value>test-bigdata1:2181,test-bigdata2:2181,test-bigdata3:2181</value>
   </property>
</configuration>

修改hadoop中etc/hadoop目录下面的hdfs-site.xml文件。

<configuration>
  <!--完全分布式集群名称 -->
  <property>
    <name>dfs.nameservices</name>
    <value>mycluster</value>
  </property>
  <!--集群中NameNode节点名称 -->
  <property>
    <name>dfs.ha.namenodes.mycluster</name>
    <value>nn1,nn2</value>
  </property>
  <!--nn1的RPC通信地址 -->
  <property>
    <name>dfs.namenode.rpc-address.mycluster.nn1</name>
    <value>test-bigdata1:9000</value>
  </property>
  <!-- nn2的RPC通信地址 -->
  <property>
    <name>dfs.namenode.rpc-address.mycluster.nn2</name>
    <value>test-bigdata2:9000</value>
  </property>
  <!--nn1的http通信地址 -->
  <property>
    <name>dfs.namenode.http-address.mycluster.nn1</name>
    <value>test-bigdata1:50070</value>
  </property>
  <!-- nn2的http通信地址 -->
  <property>
    <name>dfs.namenode.http-address.mycluster.nn2</name>
    <value>test-bigdata2:50070</value>
  </property>
  <!--指定Namenode元数据在JournalNode上的存放位置 -->
  <property>
    <name>dfs.namenode.shared.edits.dir</name>
    <value>qjournal://test-bigdata1:8485;test-bigdata2:8485;test-bigdata3:8485/mycluster</value>
  </property>
  <!--配置隔离机制，即同一时刻只能有一台服务器对外响应 -->
  <property>
    <name>dfs.ha.fencing.methods</name>
    <value>shell(/bin/true)</value>
  </property>
  <!--使用隔离机制时需要ssh无秘钥登录-->
  <property>
    <name>dfs.ha.fencing.ssh.private-key-files</name>
    <value>/root/.ssh/id_rsa</value>
  </property>
  <!--声明journalnode服务器存储目录-->
  <property>
    <name>dfs.journalnode.edits.dir</name>
    <value>/usr/local/hadoop-3.2.1/data/jn</value>
  </property>
  <!--关闭检查权限 -->
  <property>
    <name>dfs.permissions.enables</name>
    <value>false</value>
  </property>
  <!--访问代理类client，mycluster，active 配置失败自动切换实现方式-->
  <property>
    <name>dfs.client.failover.proxy.provider.mycluster</name>
    <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
  </property>
  <!-- 开启故障自动转移 -->
  <property>
    <name>dfs.ha.automatic-failover.enabled</name>
    <value>true</value>
  </property>
</configuration>

修改hadoop中etc/hadoop目录下面的yarn-site.xml文件。

<configuration>
   <property> 
       <name>yarn.nodemanager.aux-services</name>
       <value>mapreduce_shuffle</value>
   </property>
   <!--启用 resourcemanager ha--> 
   <property>
       <name>yarn.resourcemanager.ha.enabled</name>
       <value>true</value>
   </property>
    <!--声明两台 resourcemanager 的地址--> 
   <property>
       <name>yarn.resourcemanager.cluster-id</name>
       <value>cluster-yarn1</value>
   </property>
   <property>
       <name>yarn.resourcemanager.ha.rm-ids</name>
       <value>rm1,rm2</value>
   </property>
   <property>
       <name>yarn.resourcemanager.hostname.rm1</name>
       <value>test-bigdata1</value>
   </property>
   <property>
       <name>yarn.resourcemanager.hostname.rm2</name>
      <value>test-bigdata2</value>
   </property>
    <!--指定 zookeeper 集群的地址--> 
   <property>
       <name>yarn.resourcemanager.zk-address</name>
       <value>test-bigdata1:2181,test-bigdata2:2181,test-bigdata3:2181</value>
   </property>
    <!--启用自动恢复--> 
   <property>
       <name>yarn.resourcemanager.recovery.enabled</name>
       <value>true</value>
   </property>
    <!--指定 resourcemanager 的状态信息存储在 zookeeper 集群-->  
   <property>
       <name>yarn.resourcemanager.store.class</name>
       <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
   </property>
</configuration>

修改/etc/profile文件

HADOOP_HOME=/usr/local/hadoop-3.2.1
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=$HADOOP_HOME/share/hadoop/yarn/*:$HADOOP_HOME/share/hadoop/common/*:$HADOOP_HOME/share/hadoop/mapreduce/*:$HADOOP_HOME/share/hadoop/hdfs/*:$HADOOP_HOME/share/tools/*:$HADOOP_HOME/share/hadoop/httpfs/*:$HADOOP_HOME/share/hadoop/kms/*:$HADOOP_HOME/share/hadoop/common/lib/*

集群启动

启动高可用hadoop集群

启动hdfs集群。

a. 在三台服务器上分别启动journalnode。

在hadoop的bin目录下面执行下面语句
./hdfs --daemon start journalnode

b. 在配置了NameNode的任一服务器上面格式化NameNode。

# 在bigdata01服务器的hadoop的bin目录下面执行命令
./hdfs namenode -format

c. 在执行了上面命令的那台服务器上面启动NameNode。

# 在bigdata01服务器的hadoop的bin目录下面执行命令
./hdfs --daemon start namenode

d. 在另外一台配置了NameNode的服务器上面同步NameNode信息。

# 在bigdata02服务器的hadoop的bin目录下面执行命令
./hdfs namenode -bootstrapStandby

e. 格式化zk

# 在bigdata01服务器的hadoop的bin目录下面执行命令
./hdfs zkfc -formatZK

f. 启动hadoop集群

# 在bigdata01服务器的hadoop的sbin目录下面执行命令
./start-dfs.sh

g. 启动其他服务器的DataNode。

# 在bigdata02和bigdata03服务器的bin目录下面执行命令
./hdfs --daemon start datanode

启动yarn集群

a.在配置了resourcemanager的任一台服务器上面启动resourcemanager。

# 在bigdata01服务器的hadoop的sbin目录下执行命令
./start-yarn.sh

b. 在其余服务器上面启动nodemanager。

# 在bigdata02和bigdata03服务器的hadoop的bin目录下执行
./yarn --daemon start nodemanager

高可用测试

高可用测试。

hdfs高可用测试。

a.查看NameNode的状态。

# 在bigdata01和bigdata02服务器的hadoop的bin目录下面执行命令
./hdfs haadmin -getServiceState nn1
./hdfs haadmin -getServiceState nn2

b.手动切换NameNode状态

## 假如nn1的状态为standby，nn2的状态为active
# 在状态为standby的NameNode所在的机器上的hadoop的bin目录下执行命令
./hdfs haadmin -transitionToActive nn1
# 可将NameNode的状态切换为active。
# 在状态为active的NameNode所在的机器上的hadoop的bin目录下执行命令
./hdfs haadmin -transitionToStandby nn2
# 可将NameNode的状态切换为standby。

c. 将NameNode状态为active的服务器的active停掉。

# 在状态为active的NameNode所在的机器上的hadoop的bin目录下执行命令
./hdfs --daemon stop namenode
# 然后查看状态为standby的那个NameNode的状态，如果已经变为active，表示成功。

测试yarn高可用

a. 查询yarn的状态

# 在bigdata01和bigdata02服务器的hadoop的bin目录下面执行命令
./hdfs rmadmin -getServiceState rm1
./hdfs rmadmin -getServiceState rm2

b. 手动切换yarn状态

## 假如rm1的状态为standby，rm2的状态为active
# 在状态为standby的resourcemanager所在的机器上的hadoop的bin目录下执行命令
./yarn rmadmin -transitionToActive rm1
# 可将resourcemanager的状态切换为active。
# 在状态为active的Nresourcemanager所在的机器上的hadoop的bin目录下执行命令
./hdfs rmadmin -transitionToStandby rm2
# 可将resourcemanager的状态切换为standby。

c. 将resourcemanager状态为active的服务器的resourcemanager停掉。

# 在状态为active的resourcemanager所在的机器上的hadoop的bin目录下执行命令
./yarn --daemon stop resourcemanager
# 然后查看状态为standby的那个resourcemanager的状态，如果已经变为active，表示成功。

六、flink部署(YARN模式)

下载flink-1.13.5-bin-scala_2.12.tgz 。下载地址:https://archive.apache.org/dist/flink/flink-1.13.5/flink-1.13.5-bin-scala_2.12.tgz
将flink-1.13.5-bin-scala_2.12.tgz 上传服务器。
解压文件。

tar -zxf flink-1.13.5-bin-scala_2.12.tgz -C /usr/local

配置环境变量, 在/etc/profile文件中追加如下内容

export FLINK_HOME=/usr/local/flink-1.13.5/
export PATH=$PATH:$FLINK_HOME/bin:

配置集群信息，修改flink-conf.yaml，三台均修改

#修改几个地方：
jobmanager.rpc.address: test-bigdata1
rest.port: 8081
rest.address: test-bigdata1
env.hadoop.conf.dir: /usr/local/3.2.1/etc/hadoop
high-availability: zookeeper
high-availability.storageDir: hdfs://mycluster/flink/ha/
high-availability.zookeeper.quorum: test-bigdata1:2181,test-bigdata2:2181,test-bigdata3:2181

修改msaters文件

test-bigdata1:8081

修改slaves文件

test-bigdata1
test-bigdata2
test-bigdata3

添加对接的jar包到flink安装目录的lib下。

flink-connector-jdbc_2.12-1.13.1.jar
flink-connector-starrocks-1.2.2_flink-1.13_2.11.jar 
flink-shaded-hadoop2-uber-2.8.3-1.8.3.jar
mysql-connector-java-8.0.29.jar

启动集群，在一台上面启动即可。

start-cluster.sh

查看页面

http://192.168.20.164:8081/

七、安装Dinky

修改配置

下载dlink-release-0.6.7.tar.gz，下载地址为:https://github.com/DataLinkDC/dlink/releases/download/v0.6.7/dlink-release-0.6.7.tar.gz
解压到/usr/local目录下

tar -zxf dlink-release-0.6.7.tar.gz -C /usr/local
mv dlink-release-0.6.7 dlink

初始化数据库。

mysql -h127.0.0.1 -uroot -p 
# 创建账号并修改密码
CREATE user 'dlink'@'%';
alter user 'dlink'@'%' identified with mysql_native_password by '密码';
#创建数据库
create database dlink;
# 授权数据库
grant all privileges on dlink.* to "dlink"@"%";
# 刷新权限
flush privileges;
# 下面使用新创建的dlink账号登录mysql
mysql -h127.0.0.1 -udlink -p 
# 切换数据库
use dlink;
# 导入数据库
source /usr/local/dlink/sql/dlink.sql;
source /usr/local/dlink/sql/dlinkmysqlcatalog.sql;

修改配置文件。修改application.yml文件的数据库连接信息。

spring:
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/dlink?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
    username: dlink
    password: dlink
    driver-class-name: com.mysql.cj.jdbc.Driver

新增相关jar包。

#创建目录plugins
cd /usr/local/dlink/
mkdir plugins
# 将flink的lib下面的jar全都放入plugins目录下面 包含如下jar包
flink-connector-jdbc_2.12-1.13.1.jar
flink-connector-starrocks-1.2.2_flink-1.13_2.11.jar
flink-csv-1.13.5.jar
flink-dist_2.12-1.13.5.jar
flink-json-1.13.5.jar
flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
flink-shaded-zookeeper-3.4.14.jar
flink-table_2.12-1.13.5.jar
flink-table-blink_2.12-1.13.5.jar
mysql-connector-java-8.0.29.jar
# 在其下面在新增两个jar包
javax.ws.rs-api-2.0.1.jar
log4j-1.2.17.jar

进行FLink版本适配

# 查看lib目录下面的jar包，主要关注一下几个jar包
dlink-catalog-mysql-1.13-0.6.7.jar
dlink-client-1.13-0.6.7.jar
dlink-connector-jdbc-1.13-0.6.7.jar
# 查看其版本是否为1.13,如果不是,将其删除，在extends目录下找到1.13版本，将其复制到lib目录下。

新增hadoop和flink的配置文件。

# 在安装根目录下面新建hadoop_conf文件夹，将hadoop目录中的etc/hadoop目录下面的文件全都
复制到里面。
# 在安装根目录下面新建flink_conf文件夹，将flink目录中的conf目录下面的文件全都
复制到里面。

新增HDFS文件

在HDFS上面新增/dlink/jar目录。将dlink安装目录下的dlink-app-1.13-0.6.7-jar-with-dependencies.jar重命名之后传入其中。

hdfs dfs -mkdir -p /dlink/jar
cd /usr/local/dlink/jar
mv dlink-app-1.13-0.6.7-jar-with-dependencies.jar dlink-app.jar
hdfs dfs -put dlink-app.jar /dlink/jar/

在HDFS上面新增/flink/lib目录，将flink下面相关的lib和plugins全都上传到该目录下面。

hdfs dfs -mkdir -p /flink/lib
cd /usr/local/flink-1.13.5/lib
hdfs dfs -put -f flink-shaded-zookeeper-3.4.14.jar /flink/lib
/usr/local/flink-1.13.5
hdfs dfs -put -f lib /flink/lib
hdfs dfs -put -f plugins /flink/lib

启动项目

启动项目

#启动
$sh auto.sh start
#停止
$sh auto.sh stop
#重启
$sh auto.sh restart
#查看状态
$sh auto.sh status

查看

http://192.168.20.167:8888/
默认用户名/密码: admin/admin

配置相关环境

新增集群实例

在注册中心->集群管理->集群实例管理下面新增集群实例的配置信息。

配置集群配置

在注册中心->集群管理->集群配置管理下面新增集群配置的配置信息。

测试mysql2starrocks

在数据开发 -> 目录下面新建目录flink-sql。然后新建作业。选择flink-sql，然后写入如下flinksql。

-- 表都需要提前建好
-- 创建mysql连接
create table mysql_test1(
  first_classify STRING,
  second_classify STRING,
  first_variety STRING,
  second_variety STRING
)  WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://192.168.20.167:3306/inlong_test?useSSL=false',
  'table-name' = 'dim_agriculture_products_classify',
  'username' = 'test',
  'password' = 'test123456',
  'driver' = 'com.mysql.cj.jdbc.Driver'
);
-- 创建satrrocks连接
create table starrocks_test1(
  first_classify STRING,
  second_classify STRING,
  first_variety STRING,
  second_variety STRING
)  WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql://192.168.20.213:9030/dlink_test',
  'load-url' = '192.168.20.211:8050;192.168.20.212:8050;192.168.20.213:8050',
  'database-name' = 'dlink_test',
  'table-name' = 'dim_agriculture_products_classify',
  'username' = 'root',
  'password' = 'root'
);
-- 将mysql数据同步到starrocks
insert into starrocks_test1
select * from mysql_test1;

测试starrocks2mysql

在数据开发 -> 目录下面新建目录flink-sql。然后新建作业。选择flink-sql，然后写入如下flinksql

-- 表都需要提前建好
-- 创建mysql连接
create table mysql_test2(
    first_classify STRING,
    second_classify STRING,
    first_variety STRING,
    second_variety STRING
)  WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://192.168.20.167:3306/inlong_test?useSSL=false',
   'table-name' = 'dim_agriculture_products_classify_1',
   'username' = 'test',
   'password' = 'test123456',
   'driver' = 'com.mysql.cj.jdbc.Driver'
);
-- 创建satrrocks连接
create table starrocks_test2(
    first_classify STRING,
    second_classify STRING,
    first_variety STRING,
    second_variety STRING
)  WITH (
   'connector' = 'starrocks',
   'jdbc-url' = 'jdbc:mysql://192.168.20.213:9030/dlink_test',
   'scan-url' = '192.168.20.211:8050,192.168.20.212:8050,192.168.20.213:8050',
   'database-name' = 'dlink_test',
   'table-name' = 'dim_agriculture_products_classify',
   'username' = 'root',
   'password' = 'root'
);
-- 将starrocks数据导入mysql
insert into mysql_test2
select * from starrocks_test2;

点击保存 -> 检查 ->运行。

运行之后可以到flink webui查看信息。也可以到yarn上面查看任务。
测试结果

mysql表结构
CREATE TABLE `dim_agriculture_products_classify_1` (
  `first_classify` varchar(50) NULL COMMENT "一级分类",
  `second_classify` varchar(50) NULL COMMENT "二级分类",
  `first_variety` varchar(50) NULL COMMENT "一级品种",
  `second_variety` varchar(1000) NULL COMMENT "二级品种"
) 
satrrocks表结构
CREATE TABLE `dim_agriculture_products_classify` (
  `first_classify` varchar(50) NULL COMMENT "一级分类",
  `second_classify` varchar(50) NULL COMMENT "二级分类",
  `first_variety` varchar(50) NULL COMMENT "一级品种",
  `second_variety` varchar(1000) NULL COMMENT "二级品种"
) ENGINE=OLAP 
UNIQUE KEY(`first_classify`, `second_classify`, `first_variety`, `second_variety`)
COMMENT "农产品分类维度数据"
DISTRIBUTED BY HASH(`second_variety`) BUCKETS 8 
PROPERTIES (
"replication_num" = "1",
"in_memory" = "false",
"storage_format" = "DEFAULT",
"enable_persistent_index" = "false"
);
数据量1099条,
mysql -> starrocks  2s
starrocks -> mysql 5s

数据实时同步平台搭建

二、MySQL部署

三、zookeeper部署

zookeeper集群部署

部署zkui可视化页面

四、Pulsar部署

部署BookKeeper集群

部署pulsar集群

部署PulsarManager

五、HADOOP部署

修改配置文件

集群启动

高可用测试

六、flink部署(YARN模式)

七、安装Dinky

修改配置

新增HDFS文件

启动项目

配置相关环境

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

数据实时同步平台搭建

二、MySQL部署

三、zookeeper部署

zookeeper集群部署

部署zkui可视化页面

四、Pulsar部署

部署BookKeeper集群

部署pulsar集群

部署PulsarManager

五、HADOOP部署

修改配置文件

集群启动

高可用测试

六、flink部署(YARN模式)

七、安装Dinky

修改配置

新增HDFS文件

启动项目

配置相关环境

热门文章

最新文章

相关课程

相关电子书

相关实验场景