30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度1. 为什么需要关注 Doris 的部署与使用如果你正在处理海量数据的实时分析或者厌倦了传统数据仓库的复杂运维和高昂成本那么 Apache Doris 是一个值得花时间研究的选项。它不是一个新概念但近几年在实时数仓、交互式查询、报表分析这些场景里被提及的频率越来越高。很多人第一次接触 Doris是因为它宣称的“极速”和“易用”——支持标准 SQL兼容 MySQL 协议这意味着你可以用熟悉的客户端直接连接学习成本看起来很低。但真正要把它用起来特别是结合 Python 生态你会发现从“能连上”到“稳定高效地跑起来”中间有不少细节需要捋清楚。比如Doris 本身怎么部署是单机测试还是集群化Python 里用什么驱动连接参数怎么配批量导入数据时是走 Stream Load 还是 Insert这些点如果没提前规划好后期调整会很麻烦。这篇文章不会只给你一个安装命令列表而是会围绕“部署”和“使用”这两个核心动作拆解成可操作的步骤。我会先讲清楚不同部署方式的适用场景和资源要求再带你走通从 Python 连接、基础 SQL 操作到数据导入导出的完整链路。目标是让你在本地或测试环境能快速搭建一个可用的 Doris 环境并知道如何用 Python 与之交互为后续更复杂的数仓任务打好基础。2. 部署 Doris选对模式避开初期大坑部署是第一步也是最容易让人纠结的一步。Doris 支持多种部署模式选错了后面会很难受。别一上来就想着搞大规模集群根据你的目标来。2.1 明确你的部署目标开发测试 or 生产预览在动手之前先问自己三个问题用途是什么是个人学习、功能验证还是小团队的生产环境预览资源有多少可用的机器内存、CPU、磁盘空间分别是多少数据量级多大测试数据是 GB 级还是 TB 级对查询响应时间有要求吗对于绝大多数 Python 学习者或初期验证者答案通常是单机伪集群部署All-in-One。这种模式在一台机器上同时启动 FEFrontend前端和 BEBackend后端进程足够你完成所有功能测试和 Python 接口调用学习。它的优点是简单、快速资源占用相对可控建议机器内存不小于 8GB。只有当你要模拟多节点架构、测试数据分片和负载均衡或者为小规模生产做准备时才需要考虑多机集群部署。对于“Python学习【181】”这个上下文我强烈建议从单机部署开始。2.2 单机部署实操从下载到启动这里以 Linux 环境CentOS 7 或 Ubuntu 18.04为例macOS 也可参考命令略有不同。第一步环境检查与依赖安装Doris 运行需要 Java 环境。先检查java -version确保已安装 JDK 8 或 JDK 11。如果没有用包管理器安装例如在 CentOS 上sudo yum install java-11-openjdk-devel第二步下载并解压 Doris访问 Apache Doris 官网的下载页面获取最新稳定版的二进制包。例如下载 doris-2.0.4-bin.tar.gz。# 假设下载到 /opt/software 目录 cd /opt/software wget https://archive.apache.org/dist/doris/2.0.4/apache-doris-2.0.4-bin.tar.gz tar -zxvf apache-doris-2.0.4-bin.tar.gz cd apache-doris-2.0.4解压后的目录结构很重要fe目录存放前端be目录存放后端。第三步配置并启动 FEFrontend进入 FE 目录修改配置文件cd fe vi conf/fe.conf关键配置项单机测试可先保持默认或微调priority_networks: 指定 FE 绑定的 IP 网段例如192.168.1.0/24。对于单机可以注释掉或设为0.0.0.0但生产环境务必指定。meta_dir: 元数据目录默认在DORIS_HOME/doris-meta。确保该路径有写权限。初始化并启动 FE# 初始化元数据仅第一次启动需要 ./bin/start_fe.sh --daemon查看日志确认启动成功tail -f log/fe.log看到thrift server started之类的日志说明 FE 启动成功。默认 web 界面端口是8030默认查询端口是9030。第四步配置并启动 BEBackend新开一个终端进入 BE 目录cd /opt/software/apache-doris-2.0.4/be vi conf/be.conf关键配置项priority_networks: 同 FE需要与 FE 在同一网络或能互通。storage_root_path: 数据存储路径格式如storage_root_path /path/to/data;。确保磁盘空间充足且有写权限。启动 BE./bin/start_be.sh --daemon查看 BE 日志tail -f log/be.log看到heartbeat success等日志表明 BE 进程正常。第五步将 BE 节点添加到 FEFE 和 BE 是独立进程需要“认亲”。使用 MySQL 客户端或任何支持 MySQL 协议的客户端连接 FEmysql -h 127.0.0.1 -P 9030 -uroot执行以下 SQL 添加 BE 节点将your_be_host_ip替换为 BE 所在机器的 IPALTER SYSTEM ADD BACKEND your_be_host_ip:9050;9050是 BE 的心跳服务端口。添加成功后可以通过SHOW BACKENDS;命令查看 BE 状态确认Alive列为true。至此一个单机伪集群的 Doris 就跑起来了。你可以通过8030端口访问 Web UI初始无密码用户为root通过9030端口执行 SQL。2.3 部署后的关键检查点部署完不要急着跑业务先做几个检查进程是否存活jps命令应该能看到DorisFe和DorisBe进程。端口是否监听netstat -tlnp | grep -E ‘(8030|9030|9050)’查看关键端口。基础 SQL 能否执行连接9030端口执行SELECT 1;和SHOW DATABASES;。Web UI 能否访问浏览器打开http://your_fe_ip:8030。如果任何一步失败优先查看对应组件的日志文件fe/log/和be/log/下的.log文件错误信息通常很直接。3. 使用 Python 连接 Doris驱动选择与连接管理Doris 兼容 MySQL 协议这是它易用性的核心。意味着你可以用几乎所有支持 MySQL 的 Python 驱动来连接它。但不同驱动在特性支持、性能和易用性上略有差异。3.1 驱动选型PyMySQL vs. mysql-connector-python vs. SQLAlchemy对于大多数应用场景我推荐以下选择路径驱动/库优点缺点适用场景PyMySQL纯 Python 实现安装简单兼容性好。性能可能略低于 C 扩展实现。快速上手、开发测试、轻量级应用的首选。mysql-connector-pythonOracle 官方维护功能完整性能较好。安装可能稍复杂涉及 C 扩展。需要官方驱动特性、对性能有更高要求。SQLAlchemy PyMySQL提供 ORM 和连接池适合中型应用。引入额外抽象层稍重。项目已使用 SQLAlchemy或需要连接池管理等高级功能。Doris 官方 Python 客户端 (pydoris)原生支持 Doris 特有协议如 Stream Load功能最对口。生态相对较新通用性不如 MySQL 驱动。需要用到 Doris 特有数据导入方式如 Stream Load。结论对于初学者和绝大多数“使用”场景直接用 PyMySQL连接9030端口是最简单直接的。当你需要用到 Doris 特有的高效批量导入接口时再引入pydoris。安装 PyMySQLpip install PyMySQL3.2 建立连接与执行基础 SQL连接字符串的参数是关键一个常见的错误是端口或主机填错。import pymysql # 连接参数 connection_config { ‘host‘: ‘127.0.0.1‘, # FE 节点的 IP ‘port‘: 9030, # FE 的查询端口不是 8030(Web) 或 3306(默认MySQL) ‘user‘: ‘root‘, # 默认超级用户生产环境务必创建专属用户 ‘password‘: ‘‘, # 默认空密码生产环境必须修改 ‘database‘: ‘test_db‘, # 可选可以连接后再 USE database ‘charset‘: ‘utf8mb4‘, } try: # 建立连接 conn pymysql.connect(**connection_config) print(“连接成功”) # 创建游标 with conn.cursor() as cursor: # 1. 创建数据库 cursor.execute(“CREATE DATABASE IF NOT EXISTS demo_python;“) cursor.execute(“USE demo_python;“) # 2. 创建表 create_table_sql “““ CREATE TABLE IF NOT EXISTS user_behavior ( user_id INT, item_id INT, category_id INT, behavior_type VARCHAR(10), ts DATETIME ) DISTRIBUTED BY HASH(user_id) BUCKETS 10 PROPERTIES (“replication_num“ “1“); “““ cursor.execute(create_table_sql) print(“表创建成功。“) # 3. 插入数据 (小批量测试) insert_sql “INSERT INTO user_behavior VALUES (%s, %s, %s, %s, %s)“ data [ (1001, 2001, 101, ‘pv‘, ‘2024-01-01 10:00:00‘), (1001, 2002, 102, ‘cart‘, ‘2024-01-01 10:01:00‘), (1002, 2001, 101, ‘buy‘, ‘2024-01-01 10:02:00‘), ] cursor.executemany(insert_sql, data) conn.commit() # 提交事务 print(f“插入了 {cursor.rowcount} 条数据。“) # 4. 查询数据 cursor.execute(“SELECT * FROM user_behavior ORDER BY ts LIMIT 5;“) results cursor.fetchall() for row in results: print(row) except pymysql.Error as e: print(f“数据库错误: {e}“) finally: if ‘conn‘ in locals() and conn.open: conn.close() print(“连接已关闭。“)关键点说明端口是 9030这是 Doris FE 接收 MySQL 协议查询的端口不是 Web UI 的 8030。DISTRIBUTED BY HASH … BUCKETS这是 Doris 建表语法的关键用于数据分片。BUCKETS数量影响并行度测试环境可以设小点如10生产环境需要根据数据量和机器数规划。PROPERTIES (“replication_num” “1”)副本数。单机部署只能设为1集群可增加以提高数据可靠性。记得 commit()Doris 默认支持事务执行 INSERT/UPDATE/DELETE 后需要提交。3.3 连接池与长连接管理对于需要频繁执行查询的 Python 服务建议使用连接池而不是每次操作都新建连接。可以使用DBUtils或SQLAlchemy的池化功能。from dbutils.pooled_db import PooledDB import pymysql # 创建连接池 pool PooledDB( creatorpymysql, maxconnections5, # 池中最大连接数 mincached2, # 初始化时创建的空闲连接 host‘127.0.0.1‘, port9030, user‘root‘, password‘‘, database‘demo_python‘, charset‘utf8mb4‘, autocommitFalse ) # 从池中获取连接 conn pool.connection() try: with conn.cursor() as cursor: cursor.execute(“SELECT COUNT(*) FROM user_behavior;“) count cursor.fetchone()[0] print(f“总记录数: {count}“) finally: conn.close() # 将连接归还给池而非真正关闭使用连接池能有效避免频繁建立 TCP 连接的开销提升性能。4. 数据导入掌握高效写入的几种方式用 Python 向 Doris 写数据INSERT INTO只是最基础的一种。当数据量变大时你需要更高效的方案。4.1 小批量插入INSERT 语句适用于实时、小批次例如每秒几百到几千条的数据写入。import pymysql import random from datetime import datetime, timedelta def insert_small_batch(conn, batch_size100): with conn.cursor() as cursor: sql “INSERT INTO user_behavior (user_id, item_id, category_id, behavior_type, ts) VALUES (%s, %s, %s, %s, %s)“ data [] base_time datetime.now() for i in range(batch_size): user_id random.randint(1000, 9999) item_id random.randint(2000, 2999) cat_id random.randint(100, 150) behavior random.choice([‘pv‘, ‘cart‘, ‘fav‘, ‘buy‘]) ts base_time - timedelta(secondsrandom.randint(0, 3600)) data.append((user_id, item_id, cat_id, behavior, ts)) cursor.executemany(sql, data) conn.commit() print(f“批量插入 {len(data)} 条完成。“) # 使用 conn pymysql.connect(host‘127.0.0.1‘, port9030, user‘root‘, password‘‘, database‘demo_python‘) insert_small_batch(conn, 500) conn.close()注意虽然executemany是一次网络交互发送多条数据但 Doris 后端仍然是逐条处理 SQL。对于更大批量这不是最优方案。4.2 大批量导入Stream Load推荐这是 Doris 推荐的高性能批量导入方式。其原理是 Python 程序将数据转换成特定格式如 CSV、JSON通过 HTTP PUT 或 POST 直接推送到 Doris 的 BE 节点由 BE 直接写入存储层效率远高于 SQL 插入。你需要安装requests库和pydoris后者封装了 Stream Load 接口。pip install requests pydoris使用pydoris进行 Stream Loadfrom pydoris import DorisClient import pandas as pd import io # 1. 准备数据例如一个Pandas DataFrame df pd.DataFrame({ ‘user_id‘: [1003, 1004, 1005], ‘item_id‘: [3001, 3002, 3003], ‘category_id‘: [201, 202, 203], ‘behavior_type‘: [‘pv‘, ‘buy‘, ‘cart‘], ‘ts‘: [‘2024-01-01 11:00:00‘, ‘2024-01-01 11:01:00‘, ‘2024-01-01 11:02:00‘] }) # 2. 将 DataFrame 转为 CSV 格式的字节流 csv_buffer io.StringIO() df.to_csv(csv_buffer, indexFalse, headerFalse) data csv_buffer.getvalue().encode(‘utf-8‘) # 3. 配置 Doris 客户端 client DorisClient( host‘127.0.0.1‘, port8030, # 注意Stream Load 使用 FE 的 http_port默认 8030 user‘root‘, password‘‘ ) # 4. 执行 Stream Load response client.stream_load( database‘demo_python‘, table‘user_behavior‘, datadata, format‘csv‘, column_separator‘,‘, timeout30 ) # 5. 检查结果 if response[‘Status‘] ‘Success‘: print(f“Stream Load 成功导入行数: {response[‘NumberLoadedRows‘]}“) else: print(f“Stream Load 失败: {response[‘Message‘]}“)关键参数解析port8030Stream Load 走 HTTP 协议端口是 FE 的http_port默认8030不是 MySQL 端口9030。format‘csv’也支持json格式。column_separatorCSV 列分隔符。返回的response中包含状态、已加载行数、过滤行数等详细信息务必检查。4.3 其他导入方式简介Broker Load用于从 HDFS、S3 等外部存储系统导入超大规模数据。需要在 Doris 中配置 BrokerPython 端通过提交导入作业来实现。Routine Load用于持续消费 Kafka 等消息队列中的数据实现实时数据接入。这是一个常驻的导入任务在 Doris 中创建后会自动运行。Insert Into … SELECT从 Doris 的另一个表导入数据常用于内部 ETL。选择建议实时单条/小批用INSERT。定时批量导入GB级用Stream Load这是 Python 程序中最常用、最高效的批量导入方式。从外部存储导入TB级用Broker Load。持续实时流用Routine Load。5. 数据查询与导出让分析结果为你所用数据存进去最终是为了查出来、用起来。Doris 的查询能力很强这里聚焦如何用 Python 高效执行查询并处理结果。5.1 执行复杂查询与参数化避免在 Python 中拼接 SQL 字符串使用参数化查询防止 SQL 注入并处理大型结果集。import pymysql import pandas as pd def query_with_params(conn, start_date, end_date, behavior‘buy‘): sql “““ SELECT user_id, COUNT(*) as purchase_count, SUM(item_id) as total_items -- 示例聚合实际可能是金额 FROM user_behavior WHERE ts %s AND ts %s AND behavior_type %s GROUP BY user_id HAVING purchase_count 1 ORDER BY purchase_count DESC LIMIT 100 “““ with conn.cursor(pymysql.cursors.DictCursor) as cursor: # 返回字典格式 cursor.execute(sql, (start_date, end_date, behavior)) # 方式1逐行获取适合大数据量避免内存溢出 # for row in cursor: # process(row) # 方式2一次性获取所有数据量不大时 results cursor.fetchall() return results # 使用 conn pymysql.connect(host‘127.0.0.1‘, port9030, user‘root‘, password‘‘, database‘demo_python‘) data query_with_params(conn, ‘2024-01-01‘, ‘2024-01-02‘) for record in data: print(record) conn.close() # 直接转为 Pandas DataFrame 进行分析 df pd.DataFrame(data) print(df.describe())使用DictCursor可以让返回的每一行是一个字典列名作为键处理起来更直观。5.2 处理查询超时与大量数据默认情况下查询可能因为超时被中断。对于复杂查询或大数据量查询需要设置合适的超时时间。# 在连接时或游标上设置超时单位秒 conn pymysql.connect( host‘127.0.0.1‘, port9030, user‘root‘, password‘‘, database‘demo_python‘, connect_timeout10, # 连接超时 read_timeout300 # 查询读取超时长查询可设大 ) # 或者在执行 SQL 前设置会话变量Doris 特有 with conn.cursor() as cursor: cursor.execute(“SET query_timeout 300;“) # 设置本次会话查询超时为300秒 cursor.execute(“SELECT /* long_query */ ... FROM huge_table;“)对于可能返回几十万、上百万行结果的查询不要一次性fetchall()。使用fetchmany(size)或迭代游标来分批处理。with conn.cursor() as cursor: cursor.execute(“SELECT * FROM large_table;“) batch_size 5000 while True: rows cursor.fetchmany(batch_size) if not rows: break # 处理这一批 rows process_batch(rows)5.3 将查询结果导出查询结果除了在程序里使用经常需要导出为文件。可以用 Python 的标准库轻松实现。import csv def export_to_csv(conn, sql, filename): with conn.cursor() as cursor: cursor.execute(sql) # 获取列名 column_names [desc[0] for desc in cursor.description] with open(filename, ‘w‘, newline‘‘, encoding‘utf-8-sig‘) as f: writer csv.writer(f) writer.writerow(column_names) # 写入标题 # 分批写入避免内存问题 batch_size 10000 while True: rows cursor.fetchmany(batch_size) if not rows: break writer.writerows(rows) print(f“数据已导出到 {filename}“) # 使用 export_to_csv(conn, “SELECT * FROM user_behavior WHERE ts ‘2024-01-01‘“, ‘user_behavior_export.csv‘)对于超大规模导出可以考虑使用 Doris 的SELECT … INTO OUTFILE语句将结果直接导出到 Doris 所在服务器的文件系统然后再由 Python 去读取或传输这样对客户端内存压力最小。6. 集成数据可视化以 Superset 为例数据最终需要呈现。Apache Superset 是一个流行的开源 BI 工具它原生支持连接 Doris。这里简要说明如何将我们部署的 Doris 作为数据源添加到 Superset。前提条件已安装并运行 Superset可通过 Docker 或 pip 安装。已在 Superset 所在环境安装 Doris 的 Python 驱动pydoris。# 在 Superset 的运行环境中 pip install pydoris在 Superset 中添加 Doris 数据源登录 Superset Web 界面。进入Settings-Database Connections。点击 Add Database。在数据库类型中选择Apache Doris如果安装了pydoris这里会出现该选项。填写连接信息SQLAlchemy URI:doris://root:127.0.0.1:9030/demo_python格式doris://用户名:密码FE主机:查询端口/数据库本例中 root 密码为空所以root:点击Test Connection确认连接成功然后保存。之后你就可以在 Superset 中直接查询demo_python库中的表并创建丰富的图表和仪表盘了。这比在 Python 脚本中写死查询逻辑要灵活得多。7. 生产环境考量与常见问题排查当你从学习测试转向生产应用时以下几个点需要额外关注。7.1 安全与权限管理修改默认密码部署后第一件事就是修改 root 密码。SET PASSWORD FOR ‘root‘ PASSWORD(‘your_strong_password‘);创建专属用户为每个应用或业务线创建独立的数据库用户并授予最小必要权限。CREATE USER ‘app_user‘ IDENTIFIED BY ‘app_password‘; GRANT SELECT, INSERT ON demo_python.* TO ‘app_user‘;网络隔离生产环境 Doris 集群应部署在内网通过跳板机或应用服务器访问避免 FE/BE 端口直接暴露在公网。7.2 性能调优起点建表时合理分桶DISTRIBUTED BY HASH(key) BUCKETS n中的n建议设置为 BE 节点数量的整数倍如 10-20倍但单个 Bucket 数据量建议在 100MB-1GB 之间。可以先估算总数据量来反推n。利用 Rollup 物化视图对高频的聚合查询如 SUM, COUNT, MIN, MAX可以创建 Rollup 来预聚合数据极大提升查询速度。这需要在建表时或之后通过ALTER TABLE来定义。关注数据模型Doris 主要支持 Duplicate明细、Aggregate聚合、Unique主键三种模型。根据业务场景选择正确的模型是性能的基础。7.3 常见问题排查清单当你的 Python 程序连接或操作 Doris 出错时按这个顺序查连接失败检查host和port(9030) 是否正确。检查 Doris FE 进程是否存活 (jps看DorisFe)。检查防火墙是否放行了 9030 端口。用 MySQL 命令行客户端直接连一下排除 Python 驱动问题mysql -h 127.0.0.1 -P 9030 -uroot执行 SQL 报错语法错误仔细检查 SQL 语句特别是 Doris 特有的语法如DISTRIBUTED BY。表不存在确认USE database或连接时指定了正确的数据库。权限不足确认当前用户对目标表有相应操作权限。Stream Load 失败检查 FE 的http_port(默认8030) 是否可访问。检查返回的response[‘Message’]常见错误有Message: “fail to execute ingest‘: 数据格式错误检查列分隔符、换行符、列数是否与表定义匹配。Message: “tablet writer write failed‘: BE 写入失败检查 BE 日志 (be/log/be.WARNING)可能是磁盘满或副本问题。使用curl命令手动测试 Stream Load隔离 Python 代码问题echo “1,100,test“ | curl --location-trusted -u root: -T - -H “format: csv“ -H “column_separator:,“ http://127.0.0.1:8030/api/demo_python/user_behavior/_stream_load查询速度慢在 Doris Web UI (8030端口) 的Query Profile中查看慢查询的详细执行计划分析瓶颈。检查是否使用了有效的过滤条件利用了分区、分桶键。考虑为目标查询创建 Rollup 物化视图。Python 驱动相关确保pymysql或pydoris版本与 Python 版本兼容。如果使用连接池检查连接泄漏连接未归还。从部署到使用Doris 给 Python 开发者提供了一条相对平滑的路径。最关键的是理解其架构FE/BE和核心概念分桶、数据模型。在 Python 层先用 PyMySQL 搞定基础 CRUD在需要高性能批量导入时转向 Stream Load。遇到问题首先看 Doris 的 FE/BE 日志它们通常记录了最直接的错误原因。先把单机环境玩熟理解了这些基本操作和问题排查方法再去挑战集群部署和更高级的特性你会从容很多。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度