Flink - 分布式数据库

#依赖包
	https://repo1.maven.org/maven2/org/apache/flink/

---
#安装包
	https://www.apache.org/dyn/closer.lua/flink/flink-1.20.0

---
#部署
##安装JAVA11
	yum install epel-release
	yum remove java
	yum install java-11-openjdk
	yum install java-11-openjdk-devel
	
---
##下载
#####为了运行Flink，只需提前安装好 Java 11。你可以通过以下命令来检查 Java 是否已经安装正确。

java -version

---
##下载 flink-1.20.0 并解压。
	mkdir -p /opt/flink
	cd /opt/flink
	
---

https://www.apache.org/dyn/closer.lua/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz
	
---
	wget https://www.apache.org/dyn/closer.lua/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz
	
	tar -xzf flink-1.20.1-bin-scala_2.12.tgz
	cd /opt/flink-1.20.1/

---
####注意：默认只能127.0.0.1访问，如果要别的服务器访问flink，需要修改配置
	vim /opt/flink-1.20.1/conf/config.yaml

![](/media//202503/2025-03-20072713301844.png)

---
##配置checkpoints与savepoints路径

![](/media//202504/2025-04-17140153436685.png)

---
##步骤 启动集群 
	Flink 附带了一个 bash 脚本，可以用于启动本地集群。

$ ./bin/start-cluster.sh
	Starting cluster.
	Starting standalonesession daemon on host.
	Starting taskexecutor daemon on host.

#####注意：taskmanager一定要有数据

![](/media//202503/2025-03-20075443867827.png)

---
##步骤 3：提交作业（Job） 
#####Flink 提供了一个 CLI 工具 bin/flink，它可以运行打包为 Java ARchives (JAR) 的程序，并控制其执行。 提交作业(//nightlies.apache.org/flink/flink-docs-master/zh/docs/concepts/glossary/#ﬂink-job) 意味着将作业的 JAR 文件和相关依赖项上载到运行中的 Flink 集群并执行它。 并执行它。

---
#####Flink 的 Releases 附带了许多的示例作业。您可以在 examples/ 文件夹中找到。

---
	#要将字数统计作业示例部署到运行中的群集，请执行以下命令：

$ ./bin/flink run examples/streaming/WordCount.jar

---
	#您可以通过查看日志来验证输出结果：

$ tail log/flink-*-taskexecutor-*.out
	输出示例：

(nymph,1)
	  (in,3)
	  (thy,1)
	  (orisons,1)
	  (be,4)
	  (all,2)
	  (my,1)
	  (sins,1)
	  (remember,1)
	  (d,4)

![](/media//202503/2025-03-19112112714935.png)

---
##web ui验证
	http://172.16.81.205:8081/#/overview

![](/media//202503/2025-03-19112430469749.png)

---
##停止集群 
#####完成后，你可以快速停止集群和所有正在运行的组件。

$ ./bin/stop-cluster.sh
	
	
---
##依赖安装

#依赖官网
	https://repo1.maven.org/maven2/org/apache/flink/

cd /opt/flink-1.20.1/lib
	
	wget https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-doris/3.3.0/flink-cdc-pipeline-connector-doris-3.3.0.jar
	
	wget https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/3.3.0/flink-cdc-pipeline-connector-mysql-3.3.0.jar
	
	wget https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mysql-cdc/3.3.0/flink-sql-connector-mysql-cdc-3.3.0.jar

wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar
	
#####放到flink-1.20.0的lib目录下，重启服务

---
##sql client连接

CREATE TABLE mysql_ydx_network_config (
      id int,
      name varchar,
      address varchar,
      PRIMARY KEY (`id`) NOT ENFORCED
    ) WITH (
        'connector' = 'mysql-cdc',
        'hostname' = '172.16.81.206',
        'port' = '3306',
        'username' = 'root',
        'password' = 'sdfsfsdf',
        'database-name' = 'yatest',
        'table-name' = 'ydx_network_config'
    );
    
    CREATE TABLE doris_ydx_network_config (
      id int,
      name varchar,
      address varchar
    ) WITH (
        'connector' = 'doris',
        'fenodes' = '172.16.81.207:8030',
        'table.identifier' = 'yatest.ydx_network_config',
        'username' = 'root',
        'password' = 'sdfsfsdf',
        'sink.properties.format' = 'json',
        'sink.properties.read_json_by_line' = 'true'
    );
    
    set execution.checkpointing.interval = 10s;

insert into doris_ydx_network_config SELECT * FROM mysql_ydx_network_config;

![](/media//202503/2025-03-19125406865322.png)