文档说明

• 核心架构：Flink On YARN Application Mode + Paimon FileSystem Catalog + 四层数仓
• 适用场景：实时数仓、用户行为分析、实时UV/PV统计、实时业务报表
• 版本基线：JDK1.8 / Hadoop3.3.6 / ZK3.7.1 / Kafka3.2.3 / Flink1.18.0 / Paimon0.8.2 / MySQL8.0 / ClickHouse22.8

一、整体企业级架构详解

1.1 架构数据流

1.2 生产硬性规范

• 部署模式：生产环境禁止使用Session模式，统一YARN Application模式
• 元数据管理：Paimon文件目录元数据，无需Hive Metastore，元数据持久化HDFS
• 数据一致性：RocksDB状态后端 + 增量Checkpoint + Savepoint，保障Exactly-Once
• 数仓分层：严格ODS/DWD/DWS/ADS四层，职责单一、数据可追溯、可复用
• 交付规范：纯SQL文件交付，版本可控，禁止线上交互式临时运行

1.3 整体架构拓扑图

二、超详细服务器基础环境搭建

2.1 集群节点规划

2.2 全节点基础优化配置

2.2.1 关闭防火墙、SELinux（全节点执行）

systemctl stop firewalld
systemctl disable firewalld
setenforce 0
sed -i 's/SELINUX=enforcing/SELINUX=disabled/' /etc/selinux/config

2.2.2 主机名与hosts映射（全节点执行）

hostnamectl set-hostname hadoop102  # 各节点分别执行
cat >> /etc/hosts << EOF
192.168.73.205 hadoop102
192.168.73.206 hadoop103
192.168.73.207 hadoop104
EOF

2.2.3 创建专属运行用户（全节点执行）

useradd flink
passwd flink
usermod -aG wheel flink
# /opt/software  - 安装包存放目录（tar.gz/rpm包）
# /opt/bigdata   - 大数据组件安装目录（解压后的程序）
# /data/hadoop   - Hadoop NameNode/DataNode 元数据与数据块
# /data/zookeeper - ZooKeeper 快照与事务日志
# /data/kafka    - Kafka 消息日志（Segment文件）
# /data/flink    - Flink RocksDB 状态本地存储
mkdir -p /opt/software /opt/bigdata /data/hadoop /data/zookeeper /data/kafka /data/flink
chown -R flink:flink /opt/software /opt/bigdata /data/

2.2.4 配置免密登录（全节点执行）

su - flink
ssh-keygen
ssh-copy-id flink@hadoop102
ssh-copy-id flink@hadoop103
ssh-copy-id flink@hadoop104

2.2.5 全节点安装JDK1.8（全节点执行）

yum remove -y openjdk*
tar -zxvf /opt/software/jdk-8u381-linux-x64.tar.gz -C /opt/bigdata/
cat >> /etc/profile << EOF
export JAVA_HOME=/opt/bigdata/jdk1.8.0_381
export JRE_HOME=\$JAVA_HOME/jre
export CLASSPATH=.:\$JAVA_HOME/lib:\$JRE_HOME/lib
export PATH=\$JAVA_HOME/bin:\$PATH
EOF
source /etc/profile
java -version

三、大数据组件集群超详细搭建

3.1 Hadoop3.3.6 集群搭建

tar -zxvf /opt/software/hadoop-3.3.6.tar.gz -C /opt/bigdata/
chown -R flink:flink /opt/bigdata/hadoop-3.3.6

core-site.xml

<configuration>
    <property><name>fs.defaultFS</name><value>hdfs://hadoop102:9000</value></property>
    <property><name>hadoop.tmp.dir</name><value>/data/hadoop/tmp</value></property>
    <property><name>hadoop.proxyuser.flink.hosts</name><value>*</value></property>
    <property><name>hadoop.proxyuser.flink.groups</name><value>*</value></property>
</configuration>

hdfs-site.xml

<configuration>
    <property><name>dfs.replication</name><value>2</value></property>
    <property><name>dfs.namenode.secondary.http-address</name><value>hadoop103:50090</value></property>
</configuration>

yarn-site.xml

<configuration>
    <property><name>yarn.nodemanager.aux-services</name><value>mapreduce_shuffle</value></property>
    <property><name>yarn.resourcemanager.hostname</name><value>hadoop102</value></property>
    <property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value></property>
    <property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value></property>
</configuration>

workers: hadoop102 / hadoop103 / hadoop104

scp -r /opt/bigdata/hadoop-3.3.6 flink@hadoop103:/opt/bigdata/
scp -r /opt/bigdata/hadoop-3.3.6 flink@hadoop104:/opt/bigdata/
hdfs namenode -format  # 仅首次执行！
start-dfs.sh
start-yarn.sh
jps

3.2 ZooKeeper3.7.1 集群搭建

tar -zxvf /opt/software/apache-zookeeper-3.7.1-bin.tar.gz -C /opt/bigdata/
cp /opt/bigdata/apache-zookeeper-3.7.1-bin/conf/zoo_sample.cfg /opt/bigdata/apache-zookeeper-3.7.1-bin/conf/zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
clientPort=2181
server.1=hadoop102:2888:3888
server.2=hadoop103:2888:3888
server.3=hadoop104:2888:3888

echo 1 > /data/zookeeper/myid  # hadoop102
echo 2 > /data/zookeeper/myid  # hadoop103
echo 3 > /data/zookeeper/myid  # hadoop104
scp -r /opt/bigdata/apache-zookeeper-3.7.1-bin flink@hadoop103:/opt/bigdata/
scp -r /opt/bigdata/apache-zookeeper-3.7.1-bin flink@hadoop104:/opt/bigdata/
zkServer.sh start  # 所有节点
zkServer.sh status

3.3 Kafka3.2.3 集群搭建

tar -zxvf /opt/software/kafka_2.12-3.2.3.tgz -C /opt/bigdata/

broker.id=0
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://192.168.73.205:9092
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181
num.partitions=3
default.replication.factor=2
log.dirs=/data/kafka/logs

kafka-server-start.sh -daemon /opt/bigdata/kafka_2.12-3.2.3/config/server.properties

3.4 MySQL8.0 搭建

yum remove -y mariadb*
yum install -y mysql-server
systemctl start mysqld
systemctl enable mysqld
grep 'temporary password' /var/log/mysqld.log
mysql -uroot -p

ALTER USER 'root'@'localhost' IDENTIFIED BY 'YourStrongPassword@2026';
CREATE USER 'flink'@'192.168.73.%' IDENTIFIED BY 'YourStrongPassword@2026';
GRANT SELECT, INSERT, UPDATE, DELETE ON biz_db.* TO 'flink'@'192.168.73.%';
GRANT SELECT, INSERT, UPDATE, DELETE ON ads_db.* TO 'flink'@'192.168.73.%';
FLUSH PRIVILEGES;
CREATE DATABASE biz_db;
CREATE DATABASE ads_db;

3.5 ClickHouse22.8 离线部署

# 有网络的机器上提前下载 RPM 包
wget https://github.com/ClickHouse/ClickHouse/releases/download/v22.8.20.22-lts/clickhouse-server-22.8.20.22-1.el8.x86_64.rpm
wget https://github.com/ClickHouse/ClickHouse/releases/download/v22.8.20.22-lts/clickhouse-client-22.8.20.22-1.el8.x86_64.rpm
wget https://github.com/ClickHouse/ClickHouse/releases/download/v22.8.20.22-lts/clickhouse-common-static-22.8.20.22-1.el8.x86_64.rpm
# 拷贝至生产服务器后执行
yum localinstall -y /opt/software/clickhouse*.rpm
systemctl start clickhouse-server
systemctl enable clickhouse-server
clickhouse-client -q "CREATE DATABASE IF NOT EXISTS dws_db;"

四、Flink1.18.0 生产部署

tar -zxvf /opt/software/flink-1.18.0-bin-scala_2.12.tgz -C /opt/bigdata/
cat >> /etc/profile << EOF
export FLINK_HOME=/opt/bigdata/flink-1.18.0
export HADOOP_CONF_DIR=/opt/bigdata/hadoop-3.3.6/etc/hadoop
export YARN_CONF_DIR=/opt/bigdata/hadoop-3.3.6/etc/hadoop
export PATH=\$FLINK_HOME/bin:\$PATH
EOF
source /etc/profile

4.2 生产级 flink-conf.yaml

jobmanager.memory.process.size: 4g
taskmanager.memory.process.size: 8g
taskmanager.numberOfTaskSlots: 4
parallelism.default: 4
state.backend: rocksdb
state.backend.rocksdb.localdir: /data/flink/rocksdb
state.checkpoints.dir: hdfs:///flink/checkpoints
state.savepoints.dir: hdfs:///flink/savepoints
execution.checkpointing.interval: 30000
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 600000
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
state.backend.incremental: true
high-availability: zookeeper
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.storageDir: hdfs:///flink/ha
high-availability.cluster-id: /flink_prod_cluster
yarn.application-attempts: 3
yarn.maximum-failed-containers: 10
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9249

hdfs dfs -mkdir -p /flink/checkpoints /flink/savepoints /flink/ha /flink/jars /flink/paimon_warehouse
hdfs dfs -chown -R flink:flink /flink
hdfs dfs -chmod -R 755 /flink
hdfs dfs -put /opt/software/*.jar /flink/jars/

五、Paimon Catalog 元数据初始化（仅执行一次）

CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs:///flink/paimon_warehouse',
    'hadoop.conf.dir' = '/opt/bigdata/hadoop-3.3.6/etc/hadoop',
    'table-default-file-format' = 'parquet'
);
USE CATALOG paimon_catalog;
CREATE DATABASE IF NOT EXISTS ods;
CREATE DATABASE IF NOT EXISTS dwd;
CREATE DATABASE IF NOT EXISTS dws;
CREATE DATABASE IF NOT EXISTS ads;

sql-client.sh embedded -f /opt/bigdata/flink-1.18.0/sql/catalog_init.sql

六、四层实时数仓完整生产SQL

6.1 ODS层：MySQL CDC全量+增量同步

USE CATALOG paimon_catalog;
USE ods;
CREATE TABLE IF NOT EXISTS ods_user_behavior (
    user_id STRING, item_id STRING, behavior STRING,
    operate_time TIMESTAMP(3), dt STRING, hr STRING,
    PRIMARY KEY (user_id, operate_time) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc', 'hostname' = '192.168.130.145', 'port' = '3306',
    'username' = 'flink', 'password' = 'YourStrongPassword@2026',
    'database-name' = 'biz_db', 'table-name' = 'user_behavior',
    'server-id' = 5001, 'scan.startup.mode' = 'initial',
    'debezium.snapshot.locking.mode' = 'none'
);
CREATE TABLE IF NOT EXISTS ods_user_behavior_paimon (
    user_id STRING, item_id STRING, behavior STRING,
    operate_time TIMESTAMP(3), dt STRING, hr STRING
) WITH (
    'connector' = 'paimon', 'bucket' = '4', 'compression' = 'zstd',
    'write-mode' = 'append'
);
INSERT INTO ods_user_behavior_paimon
SELECT user_id, item_id, behavior, operate_time,
    DATE_FORMAT(operate_time, 'yyyy-MM-dd') AS dt,
    DATE_FORMAT(operate_time, 'HH') AS hr
FROM ods_user_behavior;

6.2 DWD层：明细清洗过滤

USE CATALOG paimon_catalog;
USE dwd;
CREATE TABLE IF NOT EXISTS dwd_user_behavior (
    user_id STRING, item_id STRING, behavior STRING,
    operate_time TIMESTAMP(3), dt STRING, hr STRING,
    PRIMARY KEY (user_id, operate_time) NOT ENFORCED
) WITH (
    'connector' = 'paimon', 'bucket' = '4', 'compression' = 'zstd',
    'write-mode' = 'merge'
);
INSERT INTO dwd_user_behavior
SELECT * FROM paimon_catalog.ods.ods_user_behavior_paimon
WHERE user_id IS NOT NULL AND item_id IS NOT NULL
  AND behavior IN ('click','cart','fav','buy');

6.3 DWS层：分钟级窗口聚合

USE CATALOG paimon_catalog;
USE dws;
CREATE TABLE IF NOT EXISTS dws_behavior_1min (
    dt STRING, hr STRING, minute STRING, behavior STRING,
    uv BIGINT, pv BIGINT, window_end TIMESTAMP(3),
    PRIMARY KEY (dt, hr, minute, behavior) NOT ENFORCED
) WITH (
    'connector' = 'clickhouse',
    'url' = 'jdbc:clickhouse://192.168.130.145:8123/dws_db',
    'table-name' = 'behavior_1min',
    'username' = 'default', 'password' = '',
    'sink.buffer-flush.interval' = '1s'
);
INSERT INTO dws_behavior_1min
SELECT dt, hr,
    DATE_FORMAT(TUMBLE_END(operate_time, INTERVAL '1' MINUTE), 'mm') AS minute,
    behavior, COUNT(DISTINCT user_id) AS uv, COUNT(item_id) AS pv,
    TUMBLE_END(operate_time, INTERVAL '1' MINUTE) AS window_end
FROM paimon_catalog.dwd.dwd_user_behavior
GROUP BY TUMBLE(operate_time, INTERVAL '1' MINUTE), dt, hr, behavior;

CREATE TABLE IF NOT EXISTS dws_db.behavior_1min (
    dt String, hr String, minute String, behavior String,
    uv UInt64, pv UInt64, window_end DateTime
) ENGINE = MergeTree()
ORDER BY (dt, hr, minute, behavior)
PARTITION BY dt
TTL window_end + INTERVAL 90 DAY;

6.4 ADS层：每日业务报表

USE CATALOG paimon_catalog;
USE ads;
CREATE TABLE IF NOT EXISTS dws_source (
    dt STRING, hr STRING, minute STRING, behavior STRING,
    uv BIGINT, pv BIGINT, window_end TIMESTAMP(3)
) WITH (
    'connector' = 'clickhouse',
    'url' = 'jdbc:clickhouse://192.168.130.145:8123/dws_db',
    'table-name' = 'behavior_1min',
    'username' = 'default', 'password' = ''
);
CREATE TABLE IF NOT EXISTS ads_daily_report (
    stat_date STRING, click_uv BIGINT, buy_uv BIGINT, total_pv BIGINT,
    update_time TIMESTAMP(3),
    PRIMARY KEY (stat_date) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.168.130.145:3306/ads_db?useSSL=false&serverTimezone=Asia/Shanghai',
    'table-name' = 'daily_report',
    'username' = 'flink', 'password' = 'YourStrongPassword@2026',
    'driver' = 'com.mysql.cj.jdbc.Driver'
);
INSERT INTO ads_daily_report
SELECT dt AS stat_date,
    SUM(CASE WHEN behavior = 'click' THEN uv ELSE 0 END) AS click_uv,
    SUM(CASE WHEN behavior = 'buy' THEN uv ELSE 0 END) AS buy_uv,
    SUM(pv) AS total_pv, CURRENT_TIMESTAMP AS update_time
FROM dws_source GROUP BY dt;

七、标准化生产作业提交脚本

#!/bin/bash
FLINK_HOME=/opt/bigdata/flink-1.18.0
HDFS_JAR_PATH=hdfs:///flink/jars
CHECKPOINT_BASE=hdfs:///flink/checkpoints
SQL_FILE=$1; JOB_NAME=$2
if [ -z "$SQL_FILE" ] || [ -z "$JOB_NAME" ]; then echo "用法: $0  "; exit 1; fi
LOG_FILE=${FLINK_HOME}/logs/${JOB_NAME}_$(date +%Y%m%d_%H%M%S).log
mkdir -p ${FLINK_HOME}/logs
nohup ${FLINK_HOME}/bin/flink run-application -t yarn-application \
    -Djobmanager.memory.process.size=4g -Dtaskmanager.memory.process.size=8g \
    -Dparallelism.default=4 -Dstate.checkpoints.dir=${CHECKPOINT_BASE}/${JOB_NAME} \
    -Dexecution.checkpointing.interval=30000 --allow-non-restored-state \
    -j ${HDFS_JAR_PATH}/flink-sql-connector-kafka_2.12-1.18.0.jar \
    -j ${HDFS_JAR_PATH}/flink-sql-connector-mysql-cdc-2.4.0.jar \
    -j ${HDFS_JAR_PATH}/flink-sql-connector-clickhouse-1.18.0.jar \
    -j ${HDFS_JAR_PATH}/flink-sql-connector-paimon-0.8.2.jar \
    -j ${HDFS_JAR_PATH}/mysql-connector-java-8.0.33.jar \
    -f ${FLINK_HOME}/sql/${SQL_FILE} > ${LOG_FILE} 2>&1 &
PID=$!; sleep 3
if kill -0 $PID 2>/dev/null; then echo "【${JOB_NAME}】提交成功"; else echo "【${JOB_NAME}】可能失败，请查日志"; tail -20 ${LOG_FILE}; exit 1; fi
chmod +x /opt/bigdata/flink-1.18.0/shell/submit_job.sh

/opt/bigdata/flink-1.18.0/shell/submit_job.sh ods_user_behavior.sql ods_job
/opt/bigdata/flink-1.18.0/shell/submit_job.sh dwd_user_behavior.sql dwd_job
/opt/bigdata/flink-1.18.0/shell/submit_job.sh dws_behavior_1min.sql dws_job
/opt/bigdata/flink-1.18.0/shell/submit_job.sh ads_daily_report.sql ads_job

八、企业级运维手册

yarn application -list | grep Flink
tail -f /opt/bigdata/flink-1.18.0/logs/ods_job_*.log
yarn application -kill <应用ID>
flink savepoint <JobID> hdfs:///flink/savepoints/

8.3 自动清理过期Checkpoint

# 仅清理Checkpoint，禁止自动清理Savepoint！
0 2 * * * hdfs dfs -find /flink/checkpoints -mtime +7 -delete

8.4 关键监控指标

九、生产规范与避坑总结

• 禁止使用 Flink Session 集群，必须 Application 模式
• Paimon Catalog 全局唯一，仅初始化一次
• 所有作业必须开启 Checkpoint，大状态开启增量快照
• 作业更新前必须手动生成 Savepoint
• 脚本、SQL 纳入 Git 版本管理
• 统一 flink 用户运行，禁止 root
• 密码禁止硬编码
• 数据库用户遵循最小权限原则

9.1 Paimon Compaction 调优

table.exec.async-compact: true
# 'num-sorted-run.compaction-trigger' = '3'
# 'compaction.min.file-num' = '3'
# 'compaction.max.file-num' = '10'

9.2 版本兼容性矩阵