开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Flink环境部署 -> 正文阅读

[大数据]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;