在实际的数据分析、报表生成和实时查询场景中，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 采用 Frontend（FE）和 Backend（BE）分离的架构，理解这一点对部署和排错至关重要。

• 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 -version

Doris 运行还需要一些系统库，建议提前安装：

sudo yum install -y epel-release sudo yum install -y wget curl telnet unzip bind-utils

2.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_port）：

ALTER 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-python

3.2 基础连接与数据库操作

下面是一个使用pymysql连接 Doris、创建数据库、创建表、插入和查询数据的完整示例。请将连接参数替换为你自己的。

# demo_basic.py import pymysql import pandas as pd # 1. 建立连接 connection = pymysql.connect( host='192.168.1.100', # FE 的 IP port=9030, # FE 的 query_port user='root', password='your_password', database='', # 初始连接可以不指定数据库 charset='utf8mb4', cursorclass=pymysql.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 特有的概念需要理解：

1. 数据模型：建表时必须指定。常见的有：

   • DUPLICATE KEY：明细模型，适合存储原始明细数据，没有主键约束，相同数据行可以重复插入。
   • AGGREGATE KEY：聚合模型，适合需要预聚合的场景（如 SUM、MAX）。插入相同 Key 的数据会自动聚合。
   • UNIQUE KEY：主键唯一模型，保证主键唯一，支持 Upsert（更新或插入）操作。 在示例中我们使用了DUPLICATE KEY，这是最通用的一种。

2. 分桶与分区：

   • DISTRIBUTED BY HASH(...) BUCKETS 10：指定数据在 BE 节点间的分布方式。HASH(user_id)表示按user_id的哈希值将数据分到 10 个桶中。分桶数建议是 BE 节点数的整数倍，单机测试可以设为一个小值（如 1-10）。
   • 还可以使用PARTITION BY RANGE(...)进行分区，常用于按时间管理数据，加速查询和简化数据生命周期管理（如删除旧分区）。

3. PROPERTIES：

   • "replication_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 作为驱动) # 格式：mysql+pymysql://user:password@host:port/database engine = create_engine('mysql+pymysql://root:your_password@192.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, index=False, if_exists='replace') 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_key=True) # 这里仅为示例，实际可能联合主键 item_id = Column(Integer, primary_key=True) category_id = Column(Integer) behavior_type = Column(String(10)) visit_time = Column(DateTime) # 创建会话 Session = sessionmaker(bind=engine) 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.csv：

user_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 = f"http://{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, headers=headers, files=files, timeout=30) # 4. 解析响应 print(f"Status Code: {response.status_code}") print(f"Response Body: {response.text}") result = response.json() if result.get('Status') == 'Success': print("Stream Load succeeded!") print(f"Loaded {result.get('NumberLoadedRows')} rows.") else: print("Stream Load failed!") print(f"Error: {result.get('Message')}") if result.get('ErrorURL'): print(f"Error 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(host='192.168.1.100', port=9030, user='root', password='your_password', database='demo_db', charset='utf8mb4') 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 连接与基础操作问题

5.2 数据导入与查询问题

5.3 生产环境部署与调优建议

学习环境可以“一键启动”，但生产环境需要考虑更多。

1. 集群规划：

   • FE：至少部署 3 个节点（1 Leader + 2 Follower）以实现高可用。Observer 节点可用于扩展读负载。
   • BE：至少 3 个节点起步，根据数据量和查询并发度水平扩展。每个 BE 节点配置应尽可能一致（CPU、内存、磁盘）。
   • 网络：所有节点需处于同一低延迟网络内，万兆网卡能显著提升性能。

2. 配置优化：

   • JVM 参数：在fe.conf和be.conf中调整JAVA_OPTS，为 FE 和 BE 分配足够堆内存。例如-Xmx8192m -Xms8192m。
   • 存储路径：storage_root_path可以配置多个逗号分隔的路径，分布在不同的磁盘上，提升 IO 能力。格式如：/path1,capacity;/path2,capacity。
   • 查询并发：通过SET global exec_mem_limit=xxx;和SET global query_timeout=xxx;等会话变量控制资源。

3. 监控与告警：

   • Doris 提供了丰富的 Metrics，可以通过 FE/BE 的 HTTP 端口（8030/8040）访问/metrics接口获取，并集成到 Prometheus + Grafana 中。
   • 重点关注 BE 节点磁盘使用率、查询延迟、导入速率、副本健康状态等指标。

4. 数据备份：

   • 定期使用BACKUP命令将数据备份到对象存储（如 S3、OSS）或 HDFS。
   • 制定数据恢复 (RESTORE) 预案并定期演练。

对于 Python 项目，在代码层面，建议将数据库连接配置外置化（如使用环境变量或配置文件），并使用连接池（如DBUtils或SQLAlchemy的QueuePool）来管理连接，避免频繁创建连接的开销。对于大批量数据写入，优先考虑使用Stream Load或Insert的多值插入，而不是逐条INSERT。

掌握 Doris 的部署和 Python 集成，为你处理大规模数据分析任务提供了一个强大而灵活的选择。从单机测试开始，理解其核心概念和操作流程，再逐步向生产集群演进，是稳妥的学习路径。接下来，你可以探索更多高级特性，如物化视图加速查询、Routine Load 从 Kafka 实时导入数据、以及利用 Bitmap 等高级聚合函数进行极速去重分析，从而更充分地释放 Doris 在实时数据分析场景下的潜力。