在实际的数据分析、报表生成和实时查询场景中MySQL 这类传统关系型数据库在处理海量数据的聚合查询时常常力不从心而 Hadoop 生态的组件又过于笨重运维复杂。Apache Doris原名 Palo作为一个基于 MPP 架构的高性能、实时的分析型数据库因其兼容 MySQL 协议、支持高并发点查和实时导入等特性成为许多数据团队构建实时数仓和 OLAP 系统的热门选择。对于 Python 开发者而言掌握 Doris 的部署与使用意味着能够将强大的数据分析能力无缝集成到自己的数据应用、自动化脚本或机器学习流水线中。本文将以一个 Python 开发者的视角带你从零开始完成一个单机版 Doris 的部署并通过 Python 代码连接 Doris、执行 SQL、处理数据最终形成一个可复现的完整流程。我们不仅会完成部署和基础操作还会深入探讨连接配置、常见错误排查以及生产环境下的最佳实践确保你学完后能独立将 Doris 应用到自己的项目中。1. 理解 Doris 的核心架构与 Python 应用场景在动手部署之前有必要先理解 Doris 的基本架构和它能为 Python 项目带来什么。这有助于你在后续配置和开发中做出正确的决策。1.1 Doris 是什么为什么选择它Apache Doris 是一个现代化的 MPP大规模并行处理分析型数据库。它的设计目标是在超大规模数据集上提供亚秒级的查询响应。与 Hive/Spark 的批处理模式不同Doris 更擅长处理高并发的实时分析查询。其核心特性包括兼容 MySQL 协议你可以使用任何支持 MySQL 协议的客户端或驱动如 Python 的mysql-connector-python、pymysql来连接 Doris学习成本极低。列式存储与向量化执行引擎针对分析查询中常见的扫描和聚合操作进行了深度优化查询速度极快。实时数据导入支持通过 Stream Load、Routine Load、Insert 等方式将 Kafka、MySQL Binlog、本地文件等数据源的数据近乎实时地导入。完善的 SQL 支持支持标准 SQL兼容 MySQL 语法包括复杂的 JOIN、窗口函数、物化视图等方便数据分析师直接使用。对于 Python 开发者Doris 的价值在于替代笨重的分析流程你可以将原本需要导出到 Pandas 再处理的大规模聚合任务直接下推给 Doris 执行大幅减少内存消耗和计算时间。构建实时数据 API利用其高并发点查能力用 Flask/FastAPI 快速构建服务于报表或应用的数据查询接口。简化数据栈一个 Doris 可以同时承担实时数仓和即席查询的角色减少了对 HBase、Presto、ClickHouse 等多个组件的依赖降低了运维复杂度。1.2 Doris 的 FE 与 BE架构简述Doris 采用 FrontendFE和 BackendBE分离的架构理解这一点对部署和排错至关重要。Frontend (FE)负责元数据管理、集群管理、查询的解析和规划。FE 节点分为 Leader、Follower 和 Observer 三种角色。用户通过 MySQL 客户端连接的就是 FE。单机部署时我们通常只启动一个 FE兼具 Leader 角色。Backend (BE)负责数据存储和查询执行。数据以 Tablet数据分片为单位存储在 BE 上查询时由 FE 协调多个 BE 并行执行。对于学习和开发测试我们可以在单台机器上同时启动 FE 和 BE。但在生产环境中它们需要部署在不同的节点上以实现高可用和水平扩展。2. 环境准备与单机版 Doris 部署我们将在一台干净的 Linux 服务器以 CentOS 7.x 为例上完成部署。请确保你拥有服务器的操作权限。2.1 系统环境检查与依赖安装首先登录你的服务器进行基础环境检查。# 1. 检查系统版本和内核 cat /etc/redhat-release uname -r # 2. 检查 Java 环境 (Doris FE 依赖 Java 8 或 11) java -version如果未安装 Java需要安装 OpenJDK# 以 CentOS 为例安装 Java 8 sudo yum install -y java-1.8.0-openjdk-devel # 验证安装 java -versionDoris 运行还需要一些系统库建议提前安装sudo yum install -y epel-release sudo yum install -y wget curl telnet unzip bind-utils2.2 下载与解压 Doris 安装包访问 Apache Doris 官网下载页 或其 GitHub Release 页面选择适合你系统的稳定版本。这里我们以2.0.5版本为例。# 创建安装目录并进入 mkdir -p /opt/doris cd /opt/doris # 下载 Doris 安装包 (请替换为最新版本的链接) wget https://archive.apache.org/dist/doris/2.0.5/apache-doris-2.0.5-bin-x64.tar.gz # 解压安装包 tar -zxvf apache-doris-2.0.5-bin-x64.tar.gz # 创建软链接或直接进入解压后的目录 ln -s apache-doris-2.0.5-bin-x64 current cd current解压后目录结构如下. ├── apache_hdfs_broker # Broker 用于访问 HDFS ├── be # Backend 目录 ├── fe # Frontend 目录 ├── udf # 用户自定义函数目录 └── www # 内置 Web 界面文件2.3 配置并启动 Frontend (FE)进入 FE 的配置目录修改核心配置文件。cd /opt/doris/current/fe cp conf/fe.conf.example conf/fe.conf vim conf/fe.conf对于单机测试你至少需要关注并修改以下配置项# 设置元数据目录确保目录存在且有写权限 meta_dir ${DORIS_HOME}/doris-meta # 设置 FE 的 IP 地址如果是单机设为当前机器内网 IP不要用 127.0.0.1 或 localhost priority_networks 192.168.1.100/24 # 请替换为你的实际网段和IP # 查询端口默认为 9030也是 MySQL 客户端连接端口 query_port 9030 # HTTP 端口用于 Web 界面和 REST API默认为 8030 http_port 8030 # 单机部署可以关闭元数据高可用检查仅用于测试 enable_meta_check false保存配置后启动 FE# 启动 FE ./bin/start_fe.sh --daemon # 查看启动日志确认是否成功 tail -f log/fe.log看到日志中出现thrift server started和start finished等字样通常表示 FE 启动成功。你也可以通过jps命令查看是否有PaloFe进程。首次启动后初始化第一次启动后需要连接到 FE 完成初始化。# 使用 MySQL 客户端连接 FE mysql -h 192.168.1.100 -P 9030 -uroot连接成功后执行以下命令设置 root 密码初始密码为空SET PASSWORD FOR root PASSWORD(your_password); ALTER USER root% IDENTIFIED BY your_password;2.4 配置并启动 Backend (BE)FE 启动成功后配置并启动 BE。cd /opt/doris/current/be cp conf/be.conf.example conf/be.conf vim conf/be.conf修改 BE 的核心配置# 设置数据存储目录确保目录存在且有写权限 storage_root_path ${DORIS_HOME}/storage # 设置 BE 的 IP 地址同样使用内网 IP priority_networks 192.168.1.100/24 # BE 的心跳服务端口用于与 FE 通信默认为 9050 heartbeat_service_port 9050 # BE 的 BRPC 端口用于 BE 间通信默认为 8060 brpc_port 8060 # BE 的 HTTP 端口用于 Web 界面默认为 8040 webserver_port 8040保存配置后启动 BE# 启动 BE ./bin/start_be.sh --daemon # 查看启动日志 tail -f log/be.log看到日志中出现heartbeat service start successfully和thrift server started通常表示 BE 启动成功。jps命令应能看到PaloBe进程。2.5 将 BE 节点添加到 Doris 集群BE 进程启动后还需要在 FE 中将其添加为集群的可用节点。# 再次使用 MySQL 客户端连接 FE mysql -h 192.168.1.100 -P 9030 -uroot -pyour_password执行以下 SQL 添加 BE 节点IP 和端口为 BE 的heartbeat_service_portALTER SYSTEM ADD BACKEND 192.168.1.100:9050;添加成功后可以通过以下命令查看 BE 状态SHOW BACKENDS\G在返回结果中关注Alive字段是否为true以及SystemDecommissioned和ClusterDecommissioned字段是否为false。这表示 BE 状态健康已成功加入集群。至此一个单机版的 Doris 集群已经部署完成。你可以通过http://192.168.1.100:8030访问 FE 的 Web 界面用户名 root密码为你设置的密码通过http://192.168.1.100:8040访问 BE 的 Web 界面。3. 使用 Python 连接与操作 Doris 数据库Doris 完全兼容 MySQL 协议因此我们可以使用任何 Python 的 MySQL 客户端库来连接。这里我们选择常用的pymysql和mysql-connector-python进行演示。3.1 准备 Python 环境与依赖在你的开发机或部署了 Doris 的服务器上需安装 Python创建虚拟环境并安装驱动。# 创建项目目录和虚拟环境 mkdir doris-python-demo cd doris-python-demo python3 -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装 MySQL 连接驱动 (二选一或都安装) pip install pymysql # 或者 pip install mysql-connector-python3.2 基础连接与数据库操作下面是一个使用pymysql连接 Doris、创建数据库、创建表、插入和查询数据的完整示例。请将连接参数替换为你自己的。# demo_basic.py import pymysql import pandas as pd # 1. 建立连接 connection pymysql.connect( host192.168.1.100, # FE 的 IP port9030, # FE 的 query_port userroot, passwordyour_password, database, # 初始连接可以不指定数据库 charsetutf8mb4, cursorclasspymysql.cursors.DictCursor # 返回字典格式的结果 ) try: with connection.cursor() as cursor: # 2. 创建数据库 create_db_sql CREATE DATABASE IF NOT EXISTS demo_db cursor.execute(create_db_sql) print(Database created or already exists.) # 3. 使用数据库 use_db_sql USE demo_db cursor.execute(use_db_sql) # 4. 创建表 (Doris 支持多种数据模型这里使用 Duplicate 模型) create_table_sql CREATE TABLE IF NOT EXISTS user_behavior ( user_id INT, item_id INT, category_id INT, behavior_type VARCHAR(10), visit_time DATETIME ) DUPLICATE KEY(user_id, item_id) DISTRIBUTED BY HASH(user_id) BUCKETS 10 PROPERTIES ( replication_num 1 ); cursor.execute(create_table_sql) print(Table user_behavior created or already exists.) # 5. 插入数据 insert_sql INSERT INTO user_behavior (user_id, item_id, category_id, behavior_type, visit_time) VALUES (1001, 2001, 1, pv, 2024-01-01 10:00:00), (1001, 2002, 2, buy, 2024-01-01 10:05:00), (1002, 2001, 1, pv, 2024-01-01 10:10:00), (1003, 2003, 3, cart, 2024-01-01 10:15:00); cursor.execute(insert_sql) connection.commit() # 提交事务 print(Data inserted successfully.) # 6. 查询数据 query_sql SELECT user_id, COUNT(*) as action_count, SUM(CASE WHEN behavior_type buy THEN 1 ELSE 0 END) as buy_count FROM user_behavior GROUP BY user_id ORDER BY action_count DESC; cursor.execute(query_sql) results cursor.fetchall() print(\nQuery Results:) for row in results: print(row) # 7. 使用 Pandas 直接读取 SQL 结果 (数据分析常用) df pd.read_sql(query_sql, connection) print(f\nDataFrame shape: {df.shape}) print(df.head()) finally: # 8. 关闭连接 connection.close()运行此脚本python demo_basic.py你将看到数据库、表被创建数据被插入并输出了聚合查询的结果。3.3 关键配置与参数详解在连接和建表时有几个 Doris 特有的概念需要理解数据模型建表时必须指定。常见的有DUPLICATE KEY明细模型适合存储原始明细数据没有主键约束相同数据行可以重复插入。AGGREGATE KEY聚合模型适合需要预聚合的场景如 SUM、MAX。插入相同 Key 的数据会自动聚合。UNIQUE KEY主键唯一模型保证主键唯一支持 Upsert更新或插入操作。 在示例中我们使用了DUPLICATE KEY这是最通用的一种。分桶与分区DISTRIBUTED BY HASH(...) BUCKETS 10指定数据在 BE 节点间的分布方式。HASH(user_id)表示按user_id的哈希值将数据分到 10 个桶中。分桶数建议是 BE 节点数的整数倍单机测试可以设为一个小值如 1-10。还可以使用PARTITION BY RANGE(...)进行分区常用于按时间管理数据加速查询和简化数据生命周期管理如删除旧分区。PROPERTIESreplication_num 1指定数据的副本数。单机部署只能设为 1。生产环境通常设为 3 以保证高可用。还可以设置storage_medium SSD、storage_cooldown_time等属性来管理数据存储。3.4 使用 SQLAlchemy 进行 ORM 操作对于使用 SQLAlchemy 框架的项目可以将其作为 MySQL 驱动来使用实现 ORM 操作。# demo_sqlalchemy.py from sqlalchemy import create_engine, Column, Integer, String, DateTime, MetaData, Table from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import pandas as pd # 创建连接引擎 (使用 pymysql 作为驱动) # 格式mysqlpymysql://user:passwordhost:port/database engine create_engine(mysqlpymysql://root:your_password192.168.1.100:9030/demo_db) # 方式一使用 Core 方式操作类似原生 SQL metadata MetaData() user_behavior_table Table(user_behavior, metadata, Column(user_id, Integer), Column(item_id, Integer), Column(category_id, Integer), Column(behavior_type, String(10)), Column(visit_time, DateTime), ) # 使用 Pandas 通过 SQLAlchemy 引擎读写数据非常方便 # 读取数据到 DataFrame df_from_db pd.read_sql_table(user_behavior, engine) print(Data read via Pandas:) print(df_from_db) # 将 DataFrame 写入新表 new_df pd.DataFrame({ user_id: [1004, 1005], item_id: [2004, 2005], category_id: [4, 5], behavior_type: [pv, fav], visit_time: [2024-01-01 11:00:00, 2024-01-01 11:05:00] }) new_df.to_sql(user_behavior_new, engine, indexFalse, if_existsreplace) print(\nData written to new table via Pandas.) # 方式二使用 ORM 声明式需注意 Doris 并非所有 SQL 特性都完全兼容 Base declarative_base() class UserBehaviorORM(Base): __tablename__ user_behavior # 注意Doris 的 Duplicate 模型表没有自增主键ORM 映射可能需要调整 user_id Column(Integer, primary_keyTrue) # 这里仅为示例实际可能联合主键 item_id Column(Integer, primary_keyTrue) category_id Column(Integer) behavior_type Column(String(10)) visit_time Column(DateTime) # 创建会话 Session sessionmaker(bindengine) session Session() # 执行查询 results session.query(UserBehaviorORM.user_id, UserBehaviorORM.behavior_type).limit(5).all() print(\nORM Query Results:) for r in results: print(r) session.close()需要注意的是由于 Doris 是分析型数据库其 SQL 语法和事务支持与 OLTP 数据库如 MySQL有差异ORM 的某些高级特性如复杂的关联关系、级联操作可能无法完美支持。通常建议在 Doris 场景下更多使用 SQL 或 Pandas 进行数据操作。4. 数据导入与查询实践连接和基础操作只是第一步Doris 的强大之处在于高效的数据导入和复杂查询。4.1 使用 Stream Load 从本地文件导入数据除了INSERT语句Doris 提供了多种高效的数据导入方式。Stream Load是一个同步的 HTTP 导入方式适合从本地文件或程序内存中导入数据。首先准备一个 CSV 文件data.csvuser_id,item_id,category_id,behavior_type,visit_time 1006,2006,6,pv,2024-01-01 12:00:00 1007,2007,7,buy,2024-01-01 12:05:00 1008,2008,8,cart,2024-01-01 12:10:00然后使用 Python 的requests库调用 Stream Load API# demo_stream_load.py import requests import base64 import json # Doris FE 的 HTTP 地址和端口 fe_host 192.168.1.100 fe_http_port 8030 db demo_db table user_behavior user root password your_password # 1. 读取本地文件内容 file_path ./data.csv with open(file_path, rb) as f: data f.read() # 2. 构建 Stream Load 请求 url fhttp://{fe_host}:{fe_http_port}/api/{db}/{table}/_stream_load headers { Authorization: Basic base64.b64encode(f{user}:{password}.encode()).decode(), Expect: 100-continue, format: csv, # 指定文件格式为 CSV column_separator: ,, # CSV 列分隔符 label: stream_load_demo_20240101, # 导入任务的唯一标签用于避免重复导入 } files {file: (data.csv, data)} # 3. 发送请求 response requests.put(url, headersheaders, filesfiles, timeout30) # 4. 解析响应 print(fStatus Code: {response.status_code}) print(fResponse Body: {response.text}) result response.json() if result.get(Status) Success: print(Stream Load succeeded!) print(fLoaded {result.get(NumberLoadedRows)} rows.) else: print(Stream Load failed!) print(fError: {result.get(Message)}) if result.get(ErrorURL): print(fError details: {result.get(ErrorURL)})Stream Load 是同步的请求返回即知道导入成功与否。label参数非常重要相同label的导入请求在短时间内只会成功一次这可以用于实现数据的“精确一次”导入语义。4.2 执行复杂分析查询Doris 支持丰富的 SQL 语法。下面演示一些分析场景中常用的查询。# demo_complex_query.py import pymysql import pandas as pd connection pymysql.connect(host192.168.1.100, port9030, userroot, passwordyour_password, databasedemo_db, charsetutf8mb4) try: with connection.cursor() as cursor: # 1. 窗口函数计算每个用户按时间的累计行为数 window_sql SELECT user_id, visit_time, behavior_type, COUNT(*) OVER (PARTITION BY user_id ORDER BY visit_time) as cumulative_actions FROM user_behavior ORDER BY user_id, visit_time; cursor.execute(window_sql) print(Window Function Results (first 5 rows):) for i in range(5): print(cursor.fetchone()) # 2. 使用 ROLLUP 进行多维聚合 rollup_sql SELECT category_id, behavior_type, COUNT(*) as count FROM user_behavior GROUP BY ROLLUP(category_id, behavior_type) ORDER BY category_id, behavior_type; df_rollup pd.read_sql(rollup_sql, connection) print(\nROLLUP Aggregation Results:) print(df_rollup) # 3. 使用 Bitmap 进行精确去重计数Doris 特色功能 # 首先创建一个支持 Bitmap 聚合的表 create_bitmap_sql CREATE TABLE IF NOT EXISTS user_distinct_demo ( dt DATE, user_id INT, city VARCHAR(50) ) AGGREGATE KEY(dt, user_id, city) DISTRIBUTED BY HASH(dt) BUCKETS 1 PROPERTIES (replication_num 1); cursor.execute(create_bitmap_sql) # 插入测试数据略 # 使用 BITMAP_UNION 和 BITMAP_UNION_COUNT 进行高效去重 bitmap_query_sql SELECT dt, BITMAP_UNION_COUNT(BITMAP_FROM_STRING(CAST(user_id AS STRING))) as distinct_users FROM user_distinct_demo GROUP BY dt; # cursor.execute(bitmap_query_sql) # 表为空时暂不执行 finally: connection.close()5. 常见问题排查与性能调优在实际使用中你可能会遇到各种问题。以下是一些典型场景的排查思路。5.1 连接与基础操作问题问题现象可能原因检查方式处理建议Python 连接失败pymysql.err.OperationalError1. FE 服务未启动或端口不对。2. 防火墙阻止了端口。3. 用户名或密码错误。4. 网络不通。1.jps查看PaloFe进程。2.telnet FE_IP 9030测试端口。3. 在服务器本地用 mysql 客户端连接测试。4. 检查 FE 配置文件fe.conf中的priority_networks。1. 启动 FE 服务。2. 开放防火墙端口9030, 8030, 9050, 8060, 8040。3. 重置密码或创建新用户。4. 确保priority_networks配置了正确的可访问 IP。SHOW BACKENDS显示 BE 状态Alive为false1. BE 服务未启动。2. FE 与 BE 网络不通。3. BE 心跳失败。1.jps查看PaloBe进程。2. 检查 BE 日志log/be.INFO有无错误。3. 在 FE 机器上telnet BE_IP 9050。1. 启动 BE 服务。2. 检查 BE 配置文件be.conf中的priority_networks。3. 查看 FE 日志log/fe.log和 BE 日志中的心跳错误信息。建表失败Failed to create table1. 语法错误。2. 分桶数BUCKETS设置不合理如远大于 BE 节点数。3. 副本数replication_num大于可用 BE 数。1. 仔细检查 SQL 语法特别是 Doris 特有的PROPERTIES部分。2. 单机测试时BUCKETS建议设为 1-10。1. 参考官方文档修正语法。2. 单机部署设置replication_num 1。3. 合理设置BUCKETS数量。5.2 数据导入与查询问题问题现象可能原因检查方式处理建议Stream Load 导入失败返回Message: “Label [xxx] already used”相同label的导入任务已经成功过Doris 拒绝了重复导入。查看返回的 JSON 信息。1. 这是正常现象保证了“至少一次”语义。如需重新导入更换一个唯一的label。2. 可以通过SHOW LOAD WHERE LABEL “xxx”;查看该标签导入的历史状态。查询速度慢1. 没有合适的索引Doris 是前缀索引。2. 数据分布严重倾斜。3. 查询没有利用分区裁剪。4. BE 节点负载高或资源不足。1. 使用EXPLAIN查看查询计划。2. 检查表的分区、分桶键选择是否合理。3. 通过 FE Web 界面或SHOW PROC ‘/backends’;查看 BE 节点负载。1. 将常用的过滤条件列放在建表语句DUPLICATE/UNIQUE/AGGREGATE KEY的前面。2. 对时间字段进行分区 (PARTITION BY RANGE)。3. 避免SELECT *只查询需要的列。4. 考虑增加 BE 节点或升级硬件。内存不足错误Memory limit exceeded查询或导入操作消耗的内存超过了单个 BE 节点的限制。查看错误日志确认是查询还是导入触发的。1. 对于查询尝试优化 SQL减少中间结果集大小。2. 对于导入调小单次导入的数据量分批进行。3. 在 BE 配置文件be.conf中调整mem_limit参数需重启 BE。5.3 生产环境部署与调优建议学习环境可以“一键启动”但生产环境需要考虑更多。集群规划FE至少部署 3 个节点1 Leader 2 Follower以实现高可用。Observer 节点可用于扩展读负载。BE至少 3 个节点起步根据数据量和查询并发度水平扩展。每个 BE 节点配置应尽可能一致CPU、内存、磁盘。网络所有节点需处于同一低延迟网络内万兆网卡能显著提升性能。配置优化JVM 参数在fe.conf和be.conf中调整JAVA_OPTS为 FE 和 BE 分配足够堆内存。例如-Xmx8192m -Xms8192m。存储路径storage_root_path可以配置多个逗号分隔的路径分布在不同的磁盘上提升 IO 能力。格式如/path1,capacity;/path2,capacity。查询并发通过SET global exec_mem_limitxxx;和SET global query_timeoutxxx;等会话变量控制资源。监控与告警Doris 提供了丰富的 Metrics可以通过 FE/BE 的 HTTP 端口8030/8040访问/metrics接口获取并集成到 Prometheus Grafana 中。重点关注 BE 节点磁盘使用率、查询延迟、导入速率、副本健康状态等指标。数据备份定期使用BACKUP命令将数据备份到对象存储如 S3、OSS或 HDFS。制定数据恢复 (RESTORE) 预案并定期演练。对于 Python 项目在代码层面建议将数据库连接配置外置化如使用环境变量或配置文件并使用连接池如DBUtils或SQLAlchemy的QueuePool来管理连接避免频繁创建连接的开销。对于大批量数据写入优先考虑使用Stream Load或Insert的多值插入而不是逐条INSERT。掌握 Doris 的部署和 Python 集成为你处理大规模数据分析任务提供了一个强大而灵活的选择。从单机测试开始理解其核心概念和操作流程再逐步向生产集群演进是稳妥的学习路径。接下来你可以探索更多高级特性如物化视图加速查询、Routine Load 从 Kafka 实时导入数据、以及利用 Bitmap 等高级聚合函数进行极速去重分析从而更充分地释放 Doris 在实时数据分析场景下的潜力。