环保网站模板 html做销售找客户的网站

环保网站模板 html,做销售找客户的网站,做网站推广见客户的话术,易语言编程可以做网站么目录 1. 环境概述 1.1 硬件与系统信息 1.2 节点规划 1.3 依赖组件 (CDH) 2. 基础环境准备 (所有节点) 2.1 检查 CPU AVX2 支持 2.2 操作系统优化 2.3 配置 Hosts 映射 2.4 创建目录与授权 3. FE (Frontend) 部署 3.1 安装与配置 3.2 启动 FE 集群 4. BE (Backend) …目录1. 环境概述1.1 硬件与系统信息1.2 节点规划1.3 依赖组件 (CDH)2. 基础环境准备 (所有节点)2.1 检查 CPU AVX2 支持2.2 操作系统优化2.3 配置 Hosts 映射2.4 创建目录与授权3. FE (Frontend) 部署3.1 安装与配置3.2 启动 FE 集群4. BE (Backend) 部署4.1 修改配置 be.conf4.2 启动 BE4.3 注册 BE5. Paimon 数据湖集成5.1 准备 Hadoop 配置文件5.2 创建 Catalog5.3 验证查询6. 问题排查与避坑总结 (Troubleshooting)7. 基础测试增删改查7.1 插入数据7.2 更新数据7.3 删除数据8. Paimon外表数据写入Doris内表基础测试8.1 数据准备8.2 测试代码8.3 数据验证1. 项目背景与技术选型当前项目正在构建基于PaimonOLAP的流批一体数仓架构。在数据链路规划中原定将治理后的数据汇入 StarRocks 【从StarRocks3.x版本开始BINARY/VARBINARY支持的最大长度与 VARCHAR 类型相同Varchar:[1, 1048576]网址BINARY/VARBINARY | StarRocks】进行查询服务。然而在实际业务场景验证中发现部分业务核心表包含超长文本或复杂 JSON 字段受限于原选型对大字段Large Field存储的支撑能力无法满足业务需求。鉴于Apache Doris 2.1.10【当时最新稳定版】在大字段存储支持 String/Text/Variant 类型【Doris的String类型默认支持 1048576 字节1MB可调大到 2147483643 字节2GB网址数据类型 - Apache Doris】以及湖仓一体Lakehouse集成方面的显著优势项目组决定调整技术选型引入 Doris 作为新的 OLAP 引擎及批处理计算单元。2. 测试目标为了验证新架构的可行性需对Doris 2.1.10 Paimon 1.1.1进行深度集成测试。本次测试的核心目标是评估 Doris 挂载 Paimon 外表的性能表现具体包括读取性能Doris 读取 Paimon ODS 层海量原始数据的速度。计算与写入性能Doris 执行 ETL 清洗逻辑数据治理并将结果回写至 Paimon DWD 层的效率。3.补充说明Paimon官网支持Doris 2.0.6 及以上版本。对应网址Doris | Apache PaimonDoirs官网支持对Paimon的查询使用 Doris 的分布式计算引擎直接访问 Paimon 数据以实现查询加速数据集成读取Paimon数据并将其写入Doris内部表或使用Doris计算引擎执行ZeroETL。对应网址Paimon Catalog - Apache Doris支持Paimon版本为1.0.0调研架构图如下1. 环境概述1.1 硬件与系统信息操作系统: CentOS 7 (CDH 6.3.2 环境混合部署)节点配置:CPU: 10核内存: 14GB (资源紧缺需精细调优)存储: 400GB SSD部署用户:bigdataJava 环境:/usr/java/jdk1.8.0_181-cloudera1.2 节点规划前置组件分配IP主机名角色版本10.x.xx.201-10.x.xx.205 10.x.xx.215 10.x.xx.149 10.x.xx.151 10.x.xx.156 10.x.xx.157 10.x.xx.167 10.x.xx.206nd1-nd5 nd6 nd11 nd12 nd13 nd14 nd15 nd16CDH6.3.210.x.xx.201-10.x.xx.205nd1-nd5Paimon1.1.1采用FE (Frontend) BE (Backend) 混合部署模式共 3 个节点。IP主机名角色端口规划 (FE/BE)备注10.x.xx.149nd11FE (Leader) BEFE: 8030, 9030, 9020, 9010 BE:18040, 9060, 9050, 8060此时 8040 被 YARN 占用BE Web 端口改为 1804010.x.xx.151nd12FE (Follower) BE同上10.x.xx.156nd13FE (Follower) BE同上1.3 依赖组件 (CDH)Hive Metastore:10.x.xx.201 (nd1),10.x.xx.203 (nd3)HDFS NameNode:10.x.xx.201 (nd1)(端口 8020)2. 基础环境准备 (所有节点)在nd11,nd12,nd13上执行以下操作。2.1 检查 CPU AVX2 支持Doris 2.0 默认依赖 AVX2 指令集。cat /proc/cpuinfo | grep avx2有输出继续下一步。无输出您需要下载 Doris 的x64-noavx2版本安装包否则 BE 启动会报错Illegal instruction。2.2 操作系统优化# 1. 临时关闭 Swap sudo swapoff -a ​ # 2. 修改 sysctl.conf sudo vi /etc/sysctl.conf # 添加 vm.max_map_count2000000 vm.swappiness0 # 生效 sudo sysctl -p# 3. 修改文件句柄限制 sudo vi /etc/security/limits.conf # 添加 * soft nofile 65536 * hard nofile 65536 * soft nproc 65536 * hard nproc 65536注意修改 limits 后需重新登录 SSH 生效可用ulimit -n检查。注意由于原始的配置均为65535Doris 官方推荐 65536 是为了取个整2的16次方但实际上 65535 对于 Doris 来说没有任何区别。只要这个数值大于 60000Doris 就能非常稳定地运行。因此上述配置可以不用进行配置。2.3 配置 Hosts 映射确保 Doris 节点能解析彼此及 CDH 组件。由于这里我是在CDH集群节点上选择的节点搭建相应的配置均有因此也可不用进行配置。sudo vi /etc/hosts ​ # Doris 节点 10.x.xx.149 nd11 10.x.xx.151 nd12 10.x.xx.156 nd13 ​ # CDH 依赖节点 10.x.xx.201 nd1 10.x.xx.203 nd32.4 创建目录与授权mkdir -p /home/bigdata/doris mkdir -p /home/bigdata/data/doris-meta # FE 元数据 mkdir -p /home/bigdata/data/doris-storage # BE 数据存储 ​ # 确保所有权为 bigdata sudo chown -R bigdata:bigdata /home/bigdata/doris sudo chown -R bigdata:bigdata /home/bigdata/data3. FE (Frontend) 部署3.1 安装与配置上传apache-doris-2.1.10-bin-x64.tar.gz至/home/bigdata/doris并解压。由于我这里选择的是nd11、nd12、nd13因此每个节点都要进行如下的配置。cd /home/bigdata/doris/apache-doris-2.1.10-bin-x64 mv fe /home/bigdata/doris/fe mv be /home/bigdata/doris/be修改配置文件fe.confvi /home/bigdata/doris/fe/conf/fe.conf关键配置项# 1. 指定 Java 环境 (解决找不到 JDK 问题) # ls -ld /usr/java JAVA_HOME/usr/java/jdk1.8.0_181-cloudera ​ # 2. 元数据目录 meta_dir /home/bigdata/data/doris-meta ​ # 3. 绑定网段 (必须指定防止抓错网卡) priority_networks 10.8.15.0/24 ​ # 4. 内存限制 (14G 内存机器FE 给 4G) JAVA_OPTS-Xmx4096m -Xms4096m -XX:UseG1GC -XX:MaxGCPauseMillis200 -XX:PrintGCDateStamps -XX:PrintGCDetails -Xloggc:log/fe.gc.log3.2 启动 FE 集群步骤 1启动 Leader (在 nd11 执行)/home/bigdata/doris/fe/bin/start_fe.sh --daemon # 停止服务 # /home/bigdata/doris/fe/bin/stop_fe.sh --daemon步骤 2启动 Follower (在 nd12, nd13 执行)注意第一次启动必须加--helper指向 Leader。/home/bigdata/doris/fe/bin/start_fe.sh --helper 10.x.xx.149:9010 --daemon查看日志确认是否启动成功tail -f /home/bigdata/doris/fe/log/fe.INFOnd11端口上述截图解析可以看出心跳广播成功 (关键信号)[topic_publish]publish topic info to be 10.x.xx.149 success ... [topic_publish]publish topic info to be 10.x.xx.151 success ... [topic_publish]publish topic info to be 10.x.xx.156 success ...含义Master FE 正在将元数据更新广播给所有的 BE 节点。状态全部success。这意味着 FE 能够顺利连接到nd11, nd12, nd13三台机器的 BE 服务。接收汇报成功receive report from be 10425 ... receive report from be 10334 ...含义BE 正在主动向 FE 汇报自己的负载和任务状态。状态正常接收。步骤 3注册 Follower在nd11上登录 MySQLmysql -h 10.x.xx.149 -P 9030 -uroot执行 SQLALTER SYSTEM ADD FOLLOWER 10.x.xx.151:9010; ALTER SYSTEM ADD FOLLOWER 10.x.xx.156:9010; -- 验证 SHOW PROC /frontends;预期结果3 个节点Alive 均为 true。4. BE (Backend) 部署4.1 修改配置be.conf所有 BE 节点需保持一致。vi /home/bigdata/doris/be/conf/be.conf关键配置项# 1. 指定 Java 环境 (Paimon/HDFS 访问必需) JAVA_HOME/usr/java/jdk1.8.0_181-cloudera # 2. 网段配置 priority_networks 10.8.15.0/24 # 3. 数据存储路径 storage_root_path /home/bigdata/data/doris-storage # 4. 修改 Web 端口 (解决 8040 被 YARN NodeManager 占用问题) webserver_port 180404.2 启动 BE在所有节点执行/home/bigdata/doris/be/bin/start_be.sh --daemon # 停止服务 # /home/bigdata/doris/be/bin/stop_be.sh --daemon查看日志确认是否启动成功tail -f /home/bigdata/doris/be/log/be.INFOnd11端口nd12端口nd13端口4.3 注册 BE在nd11 (MySQL)中执行mysql -h 10.x.xx.149 -P 9030 -uroot然后执行下述语句ALTER SYSTEM ADD BACKEND 10.x.xx.149:9050; ALTER SYSTEM ADD BACKEND 10.x.xx.151:9050; ALTER SYSTEM ADD BACKEND 10.x.xx.156:9050; -- 验证 SHOW PROC /backends;预期结果3 个节点Alive 均为 trueTotalCapacity 显示磁盘容量。5. Paimon 数据湖集成5.1 准备 Hadoop 配置文件在nd11上操作将 nd1上CDH 集群的配置文件拉取到 Doris 配置目录。mkdir -p /home/bigdata/doris/conf/cdh_conf/ # 从 CDH 节点 (201) 拷贝 # 拷贝 Hadoop 配置文件 (core-site.xml 和 hdfs-site.xml) scp root10.x.xx.201:/etc/hadoop/conf/core-site.xml /home/bigdata/doris/conf/cdh_conf/ scp root10.x.xx.201:/etc/hadoop/conf/hdfs-site.xml /home/bigdata/doris/conf/cdh_conf/ # 拷贝 Hive 配置文件 (hive-site.xml) scp root10.x.xx.201:/etc/hive/conf/hive-site.xml /home/bigdata/doris/conf/cdh_conf/验证文件ls -l /home/bigdata/doris/conf/cdh_conf/分发到其他节点 (nd12, nd13)将这 3 个文件放到 Doris 2台机器另外2台的统一目录例如/home/bigdata/doris/conf/cdh_conf/。# 1. 确保目标机器也有这个目录 ssh bigdatand12 mkdir -p /home/bigdata/doris/conf/cdh_conf/ ssh bigdatand13 mkdir -p /home/bigdata/doris/conf/cdh_conf/# 2. 发送文件给 nd12scp /home/bigdata/doris/conf/cdh_conf/* bigdatand12:/home/bigdata/doris/conf/cdh_conf/# 3. 发送文件给 nd13scp /home/bigdata/doris/conf/cdh_conf/* bigdatand13:/home/bigdata/doris/conf/cdh_conf/5.2 创建 Catalog由于 CDH 6.3.2 (Hive 2.1.1) 与 Doris 内置的高版本 Hive 客户端存在 API 兼容性问题 (报错Invalid method name: get_table_objects_by_name_req)可以使用 Filesystem 类型 Catalog【方案三】直接绕过 Hive Metastore 读取 HDFS 数据。但是为了走Hive元数据推荐使用方案四。对于API兼容性报错问题对应执行的语句为-- 方案一 CREATE CATALOG paimon_catalog PROPERTIES ( type paimon, paimon.catalog.type hms, hive.metastore.uris thrift://nd1:9083,thrift://nd3:9083, warehouse hdfs://nd1:8020/user/hive/warehouse, hadoop.conf.dir /home/bigdata/doris/conf/cdh_conf/, hadoop.username hdfs, -- 【关键修复】显式指定 Hive 版本禁止调用 Hive 3 的新 API hive.version 2.1.1 ); --方案二 --这里我还尝试了如下sql语句仍然报错Invalid method name: get_table_objects_by_name_req CREATE CATALOG paimon_catalog PROPERTIES ( type paimon, paimon.catalog.type hms, hive.metastore.uris thrift://nd1:9083,thrift://nd3:9083, warehouse hdfs://nd1:8020/user/hive/warehouse, hadoop.conf.dir /home/bigdata/doris/conf/cdh_conf/, hadoop.username hdfs, -- 【核心修改】即使你是 2.1.1也请填 1.1.0 -- 这会强制 Doris 使用旧版 API (get_table) 而不是新版 API (get_table_objects_by_name_req) hive.version 1.1.0 );上述sql语句执行成功之后会在show tables查看paimon表的时候会报错Invalid method name: get_table_objects_by_name_req。对应可以使用下述方案进行解决方案三在 Doris MySQL 客户端执行DROP CATALOG IF EXISTS paimon_catalog; CREATE CATALOG paimon_catalog PROPERTIES ( type paimon, paimon.catalog.type filesystem, -- 直接指向 HDFS 上的数仓根目录 (注意如果 nameservice 未解析直接写 active namenode 地址) warehouse hdfs://nd1:8020/user/hive/warehouse, hadoop.conf.dir /home/bigdata/doris/conf/cdh_conf/, hadoop.username hdfs );方案四在 Doris MySQL 客户端执行DROP CATALOG IF EXISTS paimon_catalog; CREATE CATALOG paimon_catalog PROPERTIES ( type paimon, -- 【这里是修改点】将 hive 改为 hms paimon.catalog.type hms, -- 指定 HMS 地址 hive.metastore.uris thrift://nd1:9083, -- 【核心兼容配置】保持不变解决 CDH 兼容性 hive.version 2.1.1, -- 数仓路径 warehouse hdfs://nd1:8020/user/hive/warehouse, -- 复用配置文件 hadoop.conf.dir /home/bigdata/doris/conf/cdh_conf/, hadoop.username hdfs );四种方案的特点总结方案核心配置 (catalog.type)HMS 地址配置hive.version结果评价方案一hms(走 Hive 元数据)双节点 HA(nd1, nd3)2.1.1失败(Invalid method)配置被忽略HA 模式下版本降级配置似乎失效了。方案二hms(走 Hive 元数据)双节点 HA(nd1, nd3)1.1.0失败(Invalid method)配置被忽略即使强制写 1.1.0HA 模式下依然顽固地用 Hive 3 协议。方案三filesystem(走文件系统)无 (直接读 HDFS)无成功绕过问题不走 Hive 协议就没有协议冲突但丧失了元数据管理能力。方案四hms(走 Hive 元数据)单节点(仅 nd1)2.1.1成功最佳实践单点模式下版本降级配置生效成功解决了兼容性问题。1.为什么方案三 (Filesystem) 可以成功原因它完全绕过了“案发现场”。原理filesystem类型的 CatalogDoris/Paimon 不会去连接 9083 端口的 Hive Metastore。它直接去 HDFS (8020 端口) 扫描目录user/hive/warehouse/ods.db/...下的 Paimon 元数据文件。为什么没报错报错的get_table_objects_by_name_req是一个 Thrift RPC 请求只有在连接 Hive Metastore 时才会发送。既然不连 HMS自然不会报错。代价这属于“降级方案”你失去了利用 HMS 进行权限控制、统一视图的能力且无法与其他 Hive 工具互通。2.为什么方案一和方案二 (HA 模式) 失败了原因Doris/Paimon 插件在处理 HA URI 列表时存在缺陷导致hive.version参数失效。现象错误Invalid method name: get_table_objects_by_name_req是典型的版本不兼容。客户端 (Doris Paimon)发出了一个 Hive 3.0 才有的“批量获取表对象”请求。服务端 (CDH Hive)我是 Hive 2.1.1我没听说过这个方法报错深层逻辑虽然你写了hive.version 2.1.1但在Scenario 1 2中你配置了多个 URI (thrift://nd1:9083,thrift://nd3:9083)。在 Doris 内部初始化 Paimon HiveCatalog 时当检测到 URI 是列表HA模式时内部的初始化逻辑可能走了一条不同的代码路径或者在传递参数时发生了丢失导致hive.version配置没有被正确应用到底层的 Hive Client。结果就是客户端“无视”了你的降级指令依然默认使用编译时的最高版本Hive 3.x去请求导致撞墙。3. 为什么方案四 (单节点 HMS) 成功了原因单点连接模式下配置参数传递正常成功触发了兼容模式。原理方案四与方案一唯一的区别就是去掉了逗号后的第二个地址。成功逻辑当hive.metastore.uris只有一个地址时Doris 正确地将hive.version 2.1.1传递给了底层的 Paimon Hive Shim。效果客户端收到了指令“对方是老版本”于是它自动禁用了get_table_objects_by_name_req这种高级 API转而使用老掉牙但兼容性好的get_tableAPI。结论这就是你一直在寻找的“既能用 HMS 管理元数据又不会报错”的完美形态。最终建议与优化既然方案四验证通过它是目前最适合的方案。关于 HA (高可用) 的补充建议如果你非常介意生产环境存在单点故障怕 nd1 挂了导致查询失败而方案一写死两个 URI又会报错。你可以尝试以下变通方法来实现 HA使用 core-site.xml 实现 Metastore 发现高级不要在 SQL 里写死 URI而是通过配置文件让 Client 自己去发现。DROP CATALOG IF EXISTS paimon_catalog; CREATE CATALOG paimon_catalog PROPERTIES ( type paimon, paimon.catalog.type hms, -- 【关键点1】注意我故意删除了 hive.metastore.uris 这一行 -- 我们赌一把让 Paimon 插件自己从下面的 xml 文件里读这个配置 -- 【关键点2】版本修复必须保留 hive.version 2.1.1, warehouse hdfs://nd1:8020/user/hive/warehouse, -- 【关键点3】这里指向你截图的文件夹必须包含 hive-site.xml hadoop.conf.dir /home/bigdata/doris/conf/cdh_conf/, hadoop.username hdfs );由截图可以得出Doris 的 Catalog 创建语法存在强校验。它强制要求在 SQL 中显式写出 hive.metastore.uris不能从配置文件“偷懒”。因此只能使用方案四它虽然在 SQL 层面只写了一个 Metastore 地址但对于解决 CDH 6.3.2 的兼容性问题它是目前唯一能打通Paimon HMS Doris的路径。运维指南由于为了兼容性牺牲了 Metastore 的自动 HA 配置建议采用以下手动 HA 方案。故障切换流程当主 Metastore 节点 (nd1) 宕机时无需删除重建 Catalog只需执行 ALTER CATALOG 命令即可毫秒级切换到备用节点 (nd3)。执行 SQL-- 将连接地址切换到 nd3 ALTER CATALOG paimon_catalog SET PROPERTIES ( hive.metastore.uris thrift://nd3:9083 );5.3 验证查询SWITCH paimon_catalog; -- 注意filesystem 模式下库名通常对应 HDFS 目录名 (可能带有 .db 后缀) SHOW DATABASES;-- ods数据库存储的为paimon表 USE ods; SHOW TABLES; SELECT * FROM t_admin_division_code LIMIT 5;6. 问题排查与避坑总结 (Troubleshooting)在本次部署过程中我们遇到了以下关键问题并成功解决问题现象报错信息关键词原因分析解决方案Java 环境缺失JAVA_HOME environment variable is not defined启动脚本未找到 CDH 自带的 JDK 路径。在fe.conf和be.conf第一行添加JAVA_HOME/usr/java/jdk1.8.0_181-cloudera。FE 启动权限错误fe.out: Permission denied或Stop it first首次使用了sudo启动导致文件归属变为 root后续bigdata用户无法写入。1. 停止 root 进程。 2.chown -R bigdata:bigdata修复目录权限。 3. 清空doris-meta和log目录。 4. 使用bigdata用户重新启动并初始化。集群无法组成System has no available disk capacity这是一个新集群的正常报错原因是 FE 启动后还未注册 BE 节点。启动 BE 并在 FE 中执行ALTER SYSTEM ADD BACKEND报错会自动消失。端口冲突tcp listen failed, errno98(端口 8040)端口 8040 被 CDH 的 YARN NodeManager 占用。修改be.conf设置webserver_port 18040。BE 启动失败 (权限)failed to create file .../.read_write_test_file: Permission deniedBE 数据存储目录 (doris-storage) 是用 root 创建的。执行sudo chown -R bigdata:bigdata /home/bigdata/data修复权限。Paimon Catalog 类型错误Unknown paimon.catalog.type value: hive-cataloghive-catalog是 Flink 的写法Doris 中应使用hms。将配置改为paimon.catalog.type hms。Hive API 版本不兼容Invalid method name: get_table_objects_by_name_reqDoris 客户端尝试使用 Hive 3.0 的批量 API但 CDH 6.3.2 (Hive 2.1.1) 不支持。放弃hms模式改用filesystem模式的 Catalog直接读取 HDFS 文件。HDFS 路径无法解析Incomplete HDFS URI, no host配置文件中未正确解析 Nameservice或未指定具体 Namenode。将warehouse路径改为明确的 NameNode 地址hdfs://nd1:8020/user/hive/warehouse。7. 基础测试增删改查将下述测试代码保存为doris2_test.py【python解释器3.8.20、windows系统11】# -*- coding: utf-8 -*- import pymysql import random import time import logging import functools from sshtunnel import SSHTunnelForwarder # 需要: pip install sshtunnel # 配置信息 # 1. SSH 连接信息 (参考你提供的 test_flink2paimon.py) SSH_HOST 10.x.xx.149 # Doris FE Leader IP (nd11) SSH_PORT 22 SSH_USER xxxxx SSH_PASSWORD xxxxxxxxxxxxxxxxx # 来自你的参考代码 # 2. Doris 数据库信息 DORIS_LOCAL_HOST 127.0.0.1 # 在服务器看来Doris是跑在本地的 DORIS_QUERY_PORT 9030 # Doris FE 查询端口 DORIS_DB_USER root DORIS_DB_PWD # 初始部署默认为空如果有设置请填写 DB_NAME python_perf_test TABLE_NAME student_scores_perf LOG_FILE doris2_test_report.log # 日志与工具模块 (保持不变) def setup_logger(): logger logging.getLogger(DorisTester) logger.setLevel(logging.INFO) if logger.hasHandlers(): logger.handlers.clear() file_handler logging.FileHandler(LOG_FILE, modew, encodingutf-8) console_handler logging.StreamHandler() formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger logger setup_logger() def measure_time(func): functools.wraps(func) def wrapper(*args, **kwargs): start_time time.time() logger.info(f正在执行: [{func.__name__}] ...) try: result func(*args, **kwargs) duration time.time() - start_time logger.info(f执行完成: [{func.__name__}] | 耗时: {duration:.4f} 秒) return result except Exception as e: duration time.time() - start_time logger.error(f执行失败: [{func.__name__}] | 耗时: {duration:.4f} 秒 | 错误: {e}) raise e return wrapper # 业务逻辑 (保持不变) measure_time def init_db_and_table(cursor): cursor.execute(fCREATE DATABASE IF NOT EXISTS {DB_NAME}) cursor.execute(fUSE {DB_NAME}) create_sql f CREATE TABLE IF NOT EXISTS {TABLE_NAME} ( id INT COMMENT 用户ID, name VARCHAR(50) COMMENT 姓名, age INT COMMENT 年龄, score INT COMMENT 分数, update_time DATETIME COMMENT 更新时间 ) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES ( replication_num 1, enable_unique_key_merge_on_write true ); # 注意测试环境副本数改为1防止BE节点不够报错 cursor.execute(create_sql) cursor.execute(fTRUNCATE TABLE {TABLE_NAME}) logger.info(f数据库 {DB_NAME} 和表 {TABLE_NAME} 已初始化) measure_time def insert_data_batch(cursor, count10): data [] for i in range(1, count 1): name fUser_{i:03d} age random.randint(18, 30) score random.randint(50, 100) data.append((i, name, age, score)) sql fINSERT INTO {TABLE_NAME} (id, name, age, score, update_time) VALUES (%s, %s, %s, %s, NOW()) cursor.executemany(sql, data) logger.info(f成功插入 {count} 条数据) measure_time def query_and_log(cursor, stage_name): sql fSELECT * FROM {TABLE_NAME} ORDER BY id cursor.execute(sql) results cursor.fetchall() logger.info(f--- [{stage_name}] 当前总行数: {len(results)} ---) if results: for row in results[:3]: logger.info(fRow: {row}) measure_time def update_random_data(cursor, update_count3): cursor.execute(fSELECT id FROM {TABLE_NAME}) all_ids [row[id] for row in cursor.fetchall()] if not all_ids: return target_ids random.sample(all_ids, min(len(all_ids), update_count)) for uid in target_ids: new_score random.randint(95, 100) sql fUPDATE {TABLE_NAME} SET score %s, update_time NOW() WHERE id %s cursor.execute(sql, (new_score, uid)) logger.info(f - 更新 ID{uid}, New Score{new_score}) measure_time def delete_random_data(cursor, delete_count2): cursor.execute(fSELECT id FROM {TABLE_NAME}) all_ids [row[id] for row in cursor.fetchall()] if not all_ids: return target_ids random.sample(all_ids, min(len(all_ids), delete_count)) for uid in target_ids: sql fDELETE FROM {TABLE_NAME} WHERE id %s cursor.execute(sql, (uid,)) logger.info(f - 删除 ID{uid}) # 主流程 (修改为使用 SSH Tunnel) def main_process(): server None conn None try: logger.info( 1. 正在建立 SSH 隧道 ...) # 建立 SSH 隧道 server SSHTunnelForwarder( (SSH_HOST, SSH_PORT), ssh_usernameSSH_USER, ssh_passwordSSH_PASSWORD, # 将远程 Doris 的 9030 映射到本地随机端口 remote_bind_address(DORIS_LOCAL_HOST, DORIS_QUERY_PORT) ) server.start() logger.info(f SSH 隧道建立成功! 本地端口: {server.local_bind_port}) # 2. 连接数据库 (连接本地端口流量会被转发) logger.info( 正在连接 Doris ...) conn pymysql.connect( host127.0.0.1, # 连接本机 portserver.local_bind_port, # 使用隧道映射的端口 userDORIS_DB_USER, passwordDORIS_DB_PWD, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, autocommitTrue ) cursor conn.cursor() # 3. 执行测试逻辑 init_db_and_table(cursor) insert_data_batch(cursor, count100) # 加大一点测试量 time.sleep(1) # Doris 写入可能有微小延迟sleep一下 query_and_log(cursor, 插入后) update_random_data(cursor, update_count5) time.sleep(1) query_and_log(cursor, 更新后) delete_random_data(cursor, delete_count5) time.sleep(1) query_and_log(cursor, 删除后) logger.info( 测试全部通过 ) except Exception as e: logger.error(f主流程发生错误: {e}) finally: # 清理资源 if conn: conn.close() logger.info(数据库连接已关闭) if server: server.stop() logger.info(SSH 隧道已关闭) if __name__ __main__: main_process()对应的log文件内容如下2025-12-09 11:21:04,416 - INFO - 1. 正在建立 SSH 隧道 ... 2025-12-09 11:21:04,766 - INFO - SSH 隧道建立成功! 本地端口: 53725 2025-12-09 11:21:04,767 - INFO - 正在连接 Doris ... 2025-12-09 11:21:04,985 - INFO - 正在执行: [init_db_and_table] ... 2025-12-09 11:21:05,251 - INFO - 数据库 python_perf_test 和表 student_scores_perf 已初始化 2025-12-09 11:21:05,251 - INFO - 执行完成: [init_db_and_table] | 耗时: 0.2653 秒 2025-12-09 11:21:05,251 - INFO - 正在执行: [insert_data_batch] ... 2025-12-09 11:21:10,981 - INFO - 成功插入 100 条数据 2025-12-09 11:21:10,984 - INFO - 执行完成: [insert_data_batch] | 耗时: 5.7328 秒 2025-12-09 11:21:11,985 - INFO - 正在执行: [query_and_log] ... 2025-12-09 11:21:12,621 - INFO - --- [插入后] 当前总行数: 100 --- 2025-12-09 11:21:12,629 - INFO - Row: {id: 1, name: User_001, age: 30, score: 52, update_time: datetime.datetime(2025, 12, 9, 11, 21, 5)} 2025-12-09 11:21:12,629 - INFO - Row: {id: 2, name: User_002, age: 25, score: 72, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:12,629 - INFO - Row: {id: 3, name: User_003, age: 28, score: 64, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:12,630 - INFO - 执行完成: [query_and_log] | 耗时: 0.6451 秒 2025-12-09 11:21:12,630 - INFO - 正在执行: [update_random_data] ... 2025-12-09 11:21:12,950 - INFO - - 更新 ID94, New Score95 2025-12-09 11:21:13,003 - INFO - - 更新 ID98, New Score98 2025-12-09 11:21:13,072 - INFO - - 更新 ID6, New Score95 2025-12-09 11:21:13,125 - INFO - - 更新 ID71, New Score97 2025-12-09 11:21:13,173 - INFO - - 更新 ID64, New Score99 2025-12-09 11:21:13,173 - INFO - 执行完成: [update_random_data] | 耗时: 0.5435 秒 2025-12-09 11:21:14,174 - INFO - 正在执行: [query_and_log] ... 2025-12-09 11:21:14,227 - INFO - --- [更新后] 当前总行数: 100 --- 2025-12-09 11:21:14,228 - INFO - Row: {id: 1, name: User_001, age: 30, score: 52, update_time: datetime.datetime(2025, 12, 9, 11, 21, 5)} 2025-12-09 11:21:14,228 - INFO - Row: {id: 2, name: User_002, age: 25, score: 72, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:14,228 - INFO - Row: {id: 3, name: User_003, age: 28, score: 64, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:14,228 - INFO - 执行完成: [query_and_log] | 耗时: 0.0545 秒 2025-12-09 11:21:14,228 - INFO - 正在执行: [delete_random_data] ... 2025-12-09 11:21:14,311 - INFO - - 删除 ID24 2025-12-09 11:21:14,361 - INFO - - 删除 ID99 2025-12-09 11:21:14,412 - INFO - - 删除 ID13 2025-12-09 11:21:14,456 - INFO - - 删除 ID22 2025-12-09 11:21:14,514 - INFO - - 删除 ID64 2025-12-09 11:21:14,514 - INFO - 执行完成: [delete_random_data] | 耗时: 0.2859 秒 2025-12-09 11:21:15,515 - INFO - 正在执行: [query_and_log] ... 2025-12-09 11:21:15,573 - INFO - --- [删除后] 当前总行数: 95 --- 2025-12-09 11:21:15,573 - INFO - Row: {id: 1, name: User_001, age: 30, score: 52, update_time: datetime.datetime(2025, 12, 9, 11, 21, 5)} 2025-12-09 11:21:15,574 - INFO - Row: {id: 2, name: User_002, age: 25, score: 72, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:15,574 - INFO - Row: {id: 3, name: User_003, age: 28, score: 64, update_time: datetime.datetime(2025, 12, 9, 11, 21, 6)} 2025-12-09 11:21:15,574 - INFO - 执行完成: [query_and_log] | 耗时: 0.0594 秒 2025-12-09 11:21:15,575 - INFO - 测试全部通过 2025-12-09 11:21:15,575 - INFO - 数据库连接已关闭 2025-12-09 11:21:15,653 - INFO - SSH 隧道已关闭去nd11执行下述语句进行查看mysql -h 10.x.xx.149 -P 9030 -urootshow databases; use python_perf_test; show tables;select * from student_scores_perf;7.1 插入数据这里展示部分数据可以看出插入了100条测试数据7.2 更新数据控制台打印结果如下终端验证部分数据展示如下7.3 删除数据控制台打印结果如下终端验证部分数据展示如下8. Paimon外表数据写入Doris内表基础测试8.1 数据准备对于Paimon外表数据写入Doris内表基础测试需要提前在Flink SQL会话里面创建Paimon表并插入测试数据-- 1. 创建 Flink 端的 Paimon Catalog CREATE CATALOG paimon_catalog WITH ( type paimon, warehouse hdfs:///user/hive/warehouse, metastore hive, hive-conf-dir /etc/hive/conf.cloudera.hive ); -- 2. 切换 Catalog 和 Database USE CATALOG my_paimon; CREATE DATABASE IF NOT EXISTS ods; USE ods; -- 3. 创建 Paimon 表 (源表) -- 这是一个记录用户行为的日志表 CREATE TABLE IF NOT EXISTS paimon_source_event ( user_id INT, item_id INT, behavior STRING, dt STRING, ts TIMESTAMP(3), PRIMARY KEY (dt, user_id, item_id) NOT ENFORCED ) PARTITIONED BY (dt) WITH ( bucket 1, file.format parquet ); -- 4. 写入测试数据 (Batch 模式写入) INSERT INTO paimon_source_event VALUES (1001, 501, click, 2025-12-12, TIMESTAMP 2025-12-12 10:00:00.123), (1002, 502, view, 2025-12-12, TIMESTAMP 2025-12-12 10:05:00.456), (1003, 501, buy, 2025-12-12, TIMESTAMP 2025-12-12 10:10:00.789), (1001, 503, view, 2025-12-13, TIMESTAMP 2025-12-13 11:00:00.000), (1004, 501, click, 2025-12-13, TIMESTAMP 2025-12-13 11:05:00.000);8.2 测试代码将下述测试代码保存为doris2_paimon_etl_test.py【python解释器3.8.20、windows系统11】# -*- coding: utf-8 -*- import pymysql import time import logging import functools from sshtunnel import SSHTunnelForwarder # 配置信息 # 1. SSH 连接信息 SSH_HOST 10.x.xx.149 SSH_PORT 22 SSH_USER xxxx SSH_PASSWORD xxxxxxxxxxxxxxxxx # 2. Doris 数据库连接信息 DORIS_LOCAL_HOST 127.0.0.1 DORIS_QUERY_PORT 9030 DORIS_DB_USER root DORIS_DB_PWD # 3. Paimon Catalog 配置 CATALOG_PROPS { type: paimon, paimon.catalog.type: hms, hive.metastore.uris: thrift://nd1:9083, hive.version: 2.1.1, warehouse: hdfs://nd1:8020/user/hive/warehouse, hadoop.conf.dir: /home/bigdata/doris/conf/cdh_conf/, hadoop.username: hdfs } # 4. 业务配置 PAIMON_CATALOG_NAME paimon_catalog PAIMON_DB ods PAIMON_TABLE paimon_source_event DORIS_TEST_DB python_perf_test DORIS_TARGET_TABLE doris_target_event_sink LOG_FILE doris_paimon_etl_report.log # 日志与工具模块 def setup_logger(): logger logging.getLogger(DorisPaimonTester) logger.setLevel(logging.INFO) if logger.hasHandlers(): logger.handlers.clear() file_handler logging.FileHandler(LOG_FILE, modew, encodingutf-8) console_handler logging.StreamHandler() formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger logger setup_logger() def measure_time(func): functools.wraps(func) def wrapper(*args, **kwargs): start_time time.time() logger.info(f正在执行: [{func.__name__}] ...) try: result func(*args, **kwargs) duration time.time() - start_time logger.info(f执行完成: [{func.__name__}] | 耗时: {duration:.4f} 秒) return result except Exception as e: duration time.time() - start_time logger.error(f执行失败: [{func.__name__}] | 耗时: {duration:.4f} 秒 | 错误: {e}) raise e return wrapper # 核心测试逻辑 measure_time def init_doris_catalog(cursor): 在 Doris 中自动创建 Paimon Catalog logger.info(f正在初始化 Doris Catalog: {PAIMON_CATALOG_NAME} ...) cursor.execute(fDROP CATALOG IF EXISTS {PAIMON_CATALOG_NAME}) props_str ,\n.join([f{k} {v} for k, v in CATALOG_PROPS.items()]) create_sql f CREATE CATALOG {PAIMON_CATALOG_NAME} PROPERTIES ( {props_str} ); cursor.execute(create_sql) logger.info(Catalog 创建成功) time.sleep(1) measure_time def check_paimon_source(cursor): 验证 Doris 是否能通过 Catalog 读取 Paimon 数据 logger.info(f检查 Paimon 数据源: {PAIMON_CATALOG_NAME}.{PAIMON_DB}.{PAIMON_TABLE}) cursor.execute(fSWITCH {PAIMON_CATALOG_NAME}) cursor.execute(fUSE {PAIMON_DB}) cursor.execute(SHOW TABLES) tables [list(row.values())[0] for row in cursor.fetchall()] if PAIMON_TABLE not in tables: raise Exception(fPaimon 表 {PAIMON_TABLE} 未找到) sql fSELECT * FROM {PAIMON_TABLE} ORDER BY dt, user_id LIMIT 5 cursor.execute(sql) results cursor.fetchall() logger.info(fPaimon 数据预览 (前5条):) for row in results: logger.info(row) if not results: raise Exception(Paimon 表为空) return len(results) measure_time def create_doris_target_table(cursor): 创建 Doris 内部表 (已修正字段顺序) cursor.execute(SWITCH internal) cursor.execute(fCREATE DATABASE IF NOT EXISTS {DORIS_TEST_DB}) cursor.execute(fUSE {DORIS_TEST_DB}) # 【注意】dt 是 Key必须紧跟在 item_id 后面 create_sql f CREATE TABLE IF NOT EXISTS {DORIS_TARGET_TABLE} ( user_id INT COMMENT 用户ID, item_id INT COMMENT 商品ID, dt VARCHAR(20) COMMENT 日期分区, behavior VARCHAR(50) COMMENT 行为类型, ts DATETIME(3) COMMENT 时间戳 ) UNIQUE KEY(user_id, item_id, dt) PARTITION BY LIST(dt) ( PARTITION p20251212 VALUES IN (2025-12-12), PARTITION p20251213 VALUES IN (2025-12-13) ) DISTRIBUTED BY HASH(user_id) BUCKETS 1 PROPERTIES ( replication_num 1, enable_unique_key_merge_on_write true ); cursor.execute(create_sql) cursor.execute(fTRUNCATE TABLE {DORIS_TARGET_TABLE}) logger.info(fDoris 内表 {DORIS_TARGET_TABLE} 已准备就绪) measure_time def execute_etl_paimon_to_doris(cursor): 执行 INSERT INTO ... SELECT ... logger.info( 开始执行从 Paimon 到 Doris 的数据导入 (ETL) ) # 显式指定插入字段顺序确保与建表顺序一致 etl_sql f INSERT INTO internal.{DORIS_TEST_DB}.{DORIS_TARGET_TABLE} (user_id, item_id, dt, behavior, ts) SELECT user_id, item_id, dt, behavior, ts FROM {PAIMON_CATALOG_NAME}.{PAIMON_DB}.{PAIMON_TABLE} cursor.execute(etl_sql) logger.info(ETL SQL 提交完毕) measure_time def verify_data_consistency(cursor): 验证两边数据是否一致 logger.info( 开始数据一致性校验 ) cursor.execute(fSELECT count(*) as cnt FROM {PAIMON_CATALOG_NAME}.{PAIMON_DB}.{PAIMON_TABLE}) paimon_count cursor.fetchone()[cnt] cursor.execute(fSELECT count(*) as cnt FROM internal.{DORIS_TEST_DB}.{DORIS_TARGET_TABLE}) doris_count cursor.fetchone()[cnt] logger.info(fPaimon 源表行数: {paimon_count}) logger.info(fDoris 目标表行数: {doris_count}) if paimon_count doris_count: logger.info(✅ 数据条数一致集成测试通过) else: logger.error(❌ 数据条数不一致。) cursor.execute(fSELECT * FROM internal.{DORIS_TEST_DB}.{DORIS_TARGET_TABLE} ORDER BY dt, user_id LIMIT 3) rows cursor.fetchall() logger.info(Doris 内表数据抽样:) for row in rows: logger.info(row) # 主流程 def main_process(): server None conn None try: logger.info( 1. 正在建立 SSH 隧道 ...) server SSHTunnelForwarder( (SSH_HOST, SSH_PORT), ssh_usernameSSH_USER, ssh_passwordSSH_PASSWORD, remote_bind_address(DORIS_LOCAL_HOST, DORIS_QUERY_PORT) ) server.start() logger.info(f SSH 隧道建立成功! 本地端口: {server.local_bind_port}) logger.info( 2. 连接 Doris ...) conn pymysql.connect( host127.0.0.1, portserver.local_bind_port, userDORIS_DB_USER, passwordDORIS_DB_PWD, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, autocommitTrue ) cursor conn.cursor() init_doris_catalog(cursor) check_paimon_source(cursor) create_doris_target_table(cursor) execute_etl_paimon_to_doris(cursor) time.sleep(2) verify_data_consistency(cursor) logger.info( 所有测试步骤执行完毕 ) except Exception as e: logger.error(f主流程发生错误: {e}) import traceback logger.error(traceback.format_exc()) finally: if conn: conn.close() if server: server.stop() if __name__ __main__: main_process()对应的log文件内容如下2025-12-12 15:19:45,098 - INFO - 1. 正在建立 SSH 隧道 ... 2025-12-12 15:19:45,395 - INFO - SSH 隧道建立成功! 本地端口: 50437 2025-12-12 15:19:45,395 - INFO - 2. 连接 Doris ... 2025-12-12 15:19:45,573 - INFO - 正在执行: [init_doris_catalog] ... 2025-12-12 15:19:45,573 - INFO - 正在初始化 Doris Catalog: paimon_catalog ... 2025-12-12 15:19:45,588 - INFO - Catalog 创建成功 2025-12-12 15:19:46,588 - INFO - 执行完成: [init_doris_catalog] | 耗时: 1.0155 秒 2025-12-12 15:19:46,589 - INFO - 正在执行: [check_paimon_source] ... 2025-12-12 15:19:46,589 - INFO - 检查 Paimon 数据源: paimon_catalog.ods.paimon_source_event 2025-12-12 15:19:46,870 - INFO - Paimon 数据预览 (前5条): 2025-12-12 15:19:46,870 - INFO - {user_id: 1001, item_id: 501, behavior: click, dt: 2025-12-12, ts: datetime.datetime(2025, 12, 12, 10, 0, 0, 123000)} 2025-12-12 15:19:46,871 - INFO - {user_id: 1002, item_id: 502, behavior: view, dt: 2025-12-12, ts: datetime.datetime(2025, 12, 12, 10, 5, 0, 456000)} 2025-12-12 15:19:46,871 - INFO - {user_id: 1003, item_id: 501, behavior: buy, dt: 2025-12-12, ts: datetime.datetime(2025, 12, 12, 10, 10, 0, 789000)} 2025-12-12 15:19:46,871 - INFO - {user_id: 1001, item_id: 503, behavior: view, dt: 2025-12-13, ts: datetime.datetime(2025, 12, 13, 11, 0)} 2025-12-12 15:19:46,871 - INFO - {user_id: 1004, item_id: 501, behavior: click, dt: 2025-12-13, ts: datetime.datetime(2025, 12, 13, 11, 5)} 2025-12-12 15:19:46,872 - INFO - 执行完成: [check_paimon_source] | 耗时: 0.2830 秒 2025-12-12 15:19:46,872 - INFO - 正在执行: [create_doris_target_table] ... 2025-12-12 15:19:46,974 - INFO - Doris 内表 doris_target_event_sink 已准备就绪 2025-12-12 15:19:46,974 - INFO - 执行完成: [create_doris_target_table] | 耗时: 0.1024 秒 2025-12-12 15:19:46,974 - INFO - 正在执行: [execute_etl_paimon_to_doris] ... 2025-12-12 15:19:46,975 - INFO - 开始执行从 Paimon 到 Doris 的数据导入 (ETL) 2025-12-12 15:19:51,286 - INFO - ETL SQL 提交完毕 2025-12-12 15:19:51,286 - INFO - 执行完成: [execute_etl_paimon_to_doris] | 耗时: 4.3116 秒 2025-12-12 15:19:53,286 - INFO - 正在执行: [verify_data_consistency] ... 2025-12-12 15:19:53,287 - INFO - 开始数据一致性校验 2025-12-12 15:19:53,579 - INFO - Paimon 源表行数: 5 2025-12-12 15:19:53,579 - INFO - Doris 目标表行数: 5 2025-12-12 15:19:53,579 - INFO - ✅ 数据条数一致集成测试通过 2025-12-12 15:19:53,605 - INFO - Doris 内表数据抽样: 2025-12-12 15:19:53,606 - INFO - {user_id: 1001, item_id: 501, dt: 2025-12-12, behavior: click, ts: datetime.datetime(2025, 12, 12, 10, 0, 0, 123000)} 2025-12-12 15:19:53,606 - INFO - {user_id: 1002, item_id: 502, dt: 2025-12-12, behavior: view, ts: datetime.datetime(2025, 12, 12, 10, 5, 0, 456000)} 2025-12-12 15:19:53,606 - INFO - {user_id: 1003, item_id: 501, dt: 2025-12-12, behavior: buy, ts: datetime.datetime(2025, 12, 12, 10, 10, 0, 789000)} 2025-12-12 15:19:53,606 - INFO - 执行完成: [verify_data_consistency] | 耗时: 0.3198 秒 2025-12-12 15:19:53,606 - INFO - 所有测试步骤执行完毕 8.3 数据验证去Doris终端验证数据结果如下mysql -h 10.x.xx.149 -P 9030 -uroot执行下述sqlSWITCH internal; SHOW DATABASES; USE python_perf_test; SHOW TABLES;SELECT * FROM doris_target_event_sink ORDER BY user_id;也可以在同一个查询窗口中直接对比两边的数量不需要反复 SWITCH-- 这里的 internal 和 paimon_catalog 是 Catalog 名称 SELECT (SELECT count(*) FROM internal.python_perf_test.doris_target_event_sink) as doris_count, (SELECT count(*) FROM paimon_catalog.ods.paimon_source_event) as paimon_count;