Flink环境部署

Posted 2022-05-02 Demonson

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flink环境部署相关的知识，希望对你有一定的参考价值。

一、基础环境搭建

1.创建用户

groupadd hadoop

useradd -g hadoop hadoop

passwd hadoop

XXX

--添加sudo权限

visudo

hadoop ALL=(ALL) NOPASSWD:ALL

2.安装需要的工具

yum install htop java lrzsz sysstat iotop -y

yum install ansible -y

3.hadoop用户下配置互信

ssh-keygen -t rsa

ssh-copy-id -i /home/hadoop/.ssh/id_rsa.pub test31

ssh-copy-id -i /home/hadoop/.ssh/id_rsa.pub hadoop@test31

4.ansible创建基础目录

--替换主机名

vim /etc/ansible/hosts

:%s/master1/test31/g

:%s/master2/test32/g

:%s/slave1/test33/g

5.ansible安装大数据组件

ansible-playbook yumtest.yaml

ansible-playbook common.yaml

ansible-playbook zookeeper.yaml

ansible-playbook scala.yaml

二、flink部署

1.上传并解压flink

cd /data/bigdata/app

rz flink-1.13.6-bin-scala_2.11.tgz

tar -zxvf flink-1.13.6-bin-scala_2.11.tgz

ln -s flink-1.13.6-bin-scala_2.11 flink

2.配置flink

vim .flink/conf/flink-conf.yaml

#配置内存大小

jobmanager.memory.process.size: 6000m

taskmanager.memory.process.size: 8000m

#配置最大任务数

taskmanager.numberOfTaskSlots: 50

#配置平行度，没有大数据量一个就可以

parallelism.default: 1

#配置checkpoint存储位置

state.backend: rocksdb

execution.checkpointing.interval: 3000

state.checkpoints.dir: hdfs://test31:9001/flink/flink-checkpoints

state.savepoints.dir: hdfs://test31:9001/flink/flink-checkpoints

#开启增量同步

state.backend.incremental: true

至此单节点flink安装完成

3.上传需要的工具包

ll flink/lib

4.启动和关闭

./bin/start-cluster.sh

./bin/stop-cluster.sh

5.通过sql-client同步任务

sql-client.sh embedded

CREATE TABLE s_user1 (


id STRING,
order_info_id STRING,
goods_id STRING,
goods_id_h STRING,
goods_name STRING,
goods_no STRING,
goods_type INT,
package_type STRING,

  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '******',
  'port' = '3306',
  'username' = '******',
  'password' = '******',
  'database-name' = 'test',
  'table-name' = 'user1',
  'server-time-zone' = 'Asia/Shanghai',
  'debezium.snapshot.mode' = 'initial'
);

CREATE TABLE user1(


id STRING,
order_info_id STRING,
goods_id STRING,
goods_id_h STRING,
goods_name STRING,
goods_no STRING,
goods_type INT,
package_type STRING,

  PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://******:3306/test?characterEncoding=UTF-8',
    'table-name' = 'user1',
    'username' = '******',
    'password' = '******'
);
  
  
INSERT INTO order_goods 
SELECT * FROM s_order_goods;