1. 项目概述在数据分析工作中我们经常需要从各种数据源提取数据进行分析。SQLite作为轻量级的嵌入式数据库因其无需服务器、零配置的特点常被用作本地数据存储方案。而Pandas作为Python数据分析的核心库提供了强大的数据处理能力。本文将详细介绍如何使用Pandas与SQLAlchemy库连接并查询SQLite数据库实现数据的高效读取和处理。这个技术组合特别适合以下场景需要快速分析存储在SQLite中的中小规模数据希望将数据库查询结果直接转换为DataFrame进行后续分析开发需要数据持久化的本地应用程序进行数据ETL(提取、转换、加载)流程的原型开发2. 环境准备与工具选型2.1 为什么选择SQLAlchemy作为连接工具SQLAlchemy是Python中最流行的ORM(对象关系映射)工具之一它提供了统一的API来操作不同类型的数据库。相比直接使用SQLite3模块SQLAlchemy具有以下优势数据库无关性同一套代码可以适配多种数据库只需修改连接字符串连接池管理自动管理数据库连接提高性能SQL注入防护内置参数化查询机制与Pandas无缝集成可以直接将查询结果转换为DataFrame2.2 安装必要的Python包在开始之前我们需要确保环境中安装了以下包conda install pandas sqlalchemy或者使用pip安装pip install pandas sqlalchemy提示建议使用conda进行安装因为它能更好地处理依赖关系特别是当项目中还涉及其他科学计算包时。3. 连接SQLite数据库3.1 创建数据库连接引擎SQLAlchemy使用引擎(Engine)对象来管理数据库连接。创建引擎的代码如下from sqlalchemy import create_engine # Windows路径示例 db_path rD:\data\store.db engine create_engine(fsqlite:///{db_path}) # Mac/Linux路径示例 # db_path /home/user/data/store.db # engine create_engine(fsqlite:////{db_path})路径说明Windows系统需要使用原始字符串(r前缀)避免转义问题SQLite连接字符串格式为sqlite:///(三个斜杠)后接数据库文件路径Linux/Mac系统需要四个斜杠因为路径以/开头3.2 连接池配置优化默认情况下SQLAlchemy会维护一个连接池。对于SQLite这种轻量级数据库我们可以调整一些参数以获得更好的性能engine create_engine( fsqlite:///{db_path}, pool_size5, # 连接池大小 max_overflow10, # 允许超出pool_size的连接数 pool_timeout30, # 获取连接的超时时间(秒) pool_recycle3600 # 连接回收时间(秒) )注意SQLite是文件型数据库过多的并发连接可能会导致性能下降甚至文件锁定问题。对于大多数应用场景保持默认配置即可。4. 使用Pandas查询数据4.1 基本查询方法Pandas提供了read_sql_query函数来执行SQL查询并直接返回DataFrameimport pandas as pd # 简单查询 query SELECT * FROM VARIABLES df pd.read_sql_query(query, engine) # 查看前10行 print(df.head(10))4.2 高级查询技巧4.2.1 参数化查询为了防止SQL注入和提高性能应该使用参数化查询# 使用命名参数 query SELECT * FROM VARIABLES WHERE category :cat df pd.read_sql_query(query, engine, params{cat: sales}) # 使用位置参数 query SELECT * FROM VARIABLES WHERE value ? df pd.read_sql_query(query, engine, params[100])4.2.2 分页查询对于大型表应该使用分页查询避免内存问题page_size 1000 page_num 1 query f SELECT * FROM VARIABLES LIMIT {page_size} OFFSET {(page_num-1)*page_size} df pd.read_sql_query(query, engine)4.2.3 使用索引加速查询如果经常按某列查询可以先为该列创建索引# 创建索引(只需执行一次) with engine.connect() as conn: conn.execute(CREATE INDEX IF NOT EXISTS idx_category ON VARIABLES(category)) # 使用索引列查询 query SELECT * FROM VARIABLES WHERE category sales df pd.read_sql_query(query, engine)5. 数据类型处理5.1 SQLite与Pandas类型映射SQLite和Pandas的数据类型不完全一致需要注意以下映射关系SQLite类型Pandas类型说明INTEGERint64REALfloat64TEXTobject字符串BLOBobject二进制数据NULLNaN缺失值5.2 处理日期时间数据SQLite没有专门的日期类型通常以TEXT、INTEGER或REAL存储。Pandas可以自动转换常见格式# 假设date列存储为YYYY-MM-DD格式 query SELECT * FROM EVENTS df pd.read_sql_query(query, engine, parse_dates[date]) # 自定义日期格式 df[date] pd.to_datetime(df[date], format%Y/%m/%d)5.3 处理缺失值SQLite中的NULL值会被转换为Pandas的NaN。我们可以使用以下方法处理# 检查缺失值 print(df.isnull().sum()) # 填充缺失值 df.fillna({column1: 0, column2: unknown}, inplaceTrue) # 删除包含缺失值的行 df.dropna(subset[important_column], inplaceTrue)6. 性能优化技巧6.1 只查询需要的列避免使用SELECT *只查询必要的列query SELECT id, name, value FROM VARIABLES df pd.read_sql_query(query, engine)6.2 使用chunksize处理大数据对于大型表可以使用chunksize参数分批读取chunk_iter pd.read_sql_query(SELECT * FROM LARGE_TABLE, engine, chunksize10000) for chunk in chunk_iter: process(chunk) # 处理每个数据块6.3 使用索引加速查询确保查询条件使用索引列# 好的查询(使用索引列) query SELECT * FROM VARIABLES WHERE indexed_column ? # 不好的查询(全表扫描) query SELECT * FROM VARIABLES WHERE non_indexed_column ?6.4 批量插入数据使用Pandas的to_sql方法批量插入数据# 准备数据 new_data pd.DataFrame({ name: [A, B, C], value: [10, 20, 30] }) # 批量插入 new_data.to_sql(VARIABLES, engine, if_existsappend, indexFalse)参数说明if_existsappend追加数据而不是覆盖表indexFalse不写入DataFrame的索引列7. 常见问题与解决方案7.1 数据库锁定问题当多个进程同时访问SQLite数据库时可能会遇到database is locked错误。解决方案确保及时关闭连接with engine.connect() as conn: df pd.read_sql_query(query, conn)设置较长的超时时间engine create_engine(fsqlite:///{db_path}, connect_args{timeout: 30})使用WAL(Write-Ahead Logging)模式engine create_engine(fsqlite:///{db_path}, connect_args{timeout: 30, isolation_level: IMMEDIATE}) with engine.connect() as conn: conn.execute(PRAGMA journal_modeWAL)7.2 内存不足问题处理大型数据库时可能遇到内存不足。解决方法使用chunksize分批读取只查询需要的列在SQL端进行聚合过滤减少返回数据量7.3 数据类型转换问题当SQLite中的数据类型与Pandas预期不符时显式指定数据类型dtype { id: int64, name: str, value: float64 } df pd.read_sql_query(query, engine, dtypedtype)查询后转换类型df[column] df[column].astype(int64)7.4 路径问题在不同操作系统上部署时路径处理要小心from pathlib import Path db_path Path(data) / store.db # 跨平台路径构造 engine create_engine(fsqlite:///{db_path.absolute()})8. 实际应用案例8.1 数据分析工作流示例# 1. 连接数据库 engine create_engine(sqlite:///sales.db) # 2. 查询销售数据 query SELECT strftime(%Y-%m, sale_date) AS month, product_category, SUM(amount) AS total_sales FROM sales WHERE sale_date 2023-01-01 GROUP BY month, product_category ORDER BY month, total_sales DESC sales_df pd.read_sql_query(query, engine) # 3. 数据分析 top_categories sales_df.groupby(month).apply( lambda x: x.nlargest(3, total_sales) ) # 4. 可视化 import matplotlib.pyplot as plt for month, data in top_categories.groupby(month): data.plot.bar(xproduct_category, ytotal_sales, titlemonth) plt.show()8.2 数据清洗与转换# 从数据库读取原始数据 raw_df pd.read_sql_query(SELECT * FROM customer_data, engine) # 数据清洗 clean_df raw_df.copy() clean_df[email] clean_df[email].str.lower().str.strip() clean_df[phone] clean_df[phone].str.replace(r\D, , regexTrue) # 处理缺失值 clean_df[segment] clean_df[segment].fillna(unknown) # 保存清洗后的数据 clean_df.to_sql(clean_customer_data, engine, if_existsreplace, indexFalse)9. 高级技巧与最佳实践9.1 使用SQLAlchemy Core进行复杂查询对于复杂查询可以结合SQLAlchemy Core使用from sqlalchemy import Table, MetaData, select metadata MetaData() variables Table(VARIABLES, metadata, autoload_withengine) # 构建查询 query select( variables.c.id, variables.c.name ).where( variables.c.value 100 ).order_by( variables.c.name ) # 执行查询 df pd.read_sql_query(query, engine)9.2 事务管理确保数据一致性使用事务操作with engine.begin() as conn: # 查询数据 df pd.read_sql_query(SELECT * FROM ACCOUNTS WHERE balance 1000, conn) # 更新数据 conn.execute( UPDATE ACCOUNTS SET status reviewed WHERE balance 1000 )9.3 使用连接上下文管理器始终使用连接上下文管理器确保资源释放with engine.connect() as conn: df1 pd.read_sql_query(SELECT * FROM TABLE1, conn) df2 pd.read_sql_query(SELECT * FROM TABLE2, conn) # 可以在同一个连接上执行多个查询 result conn.execute(SELECT COUNT(*) FROM TABLE1) count result.scalar()9.4 性能监控与调优使用SQLAlchemy的事件系统监控查询性能from sqlalchemy import event import time event.listens_for(engine, before_cursor_execute) def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): context._query_start_time time.time() event.listens_for(engine, after_cursor_execute) def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): duration time.time() - context._query_start_time if duration 0.5: # 记录慢查询 print(fSlow query ({duration:.2f}s): {statement})10. 安全注意事项10.1 防止SQL注入永远不要直接拼接SQL查询字符串# 错误做法(危险!) user_input admin; DROP TABLE users;-- query fSELECT * FROM users WHERE username {user_input} # 正确做法(使用参数化查询) query SELECT * FROM users WHERE username ? df pd.read_sql_query(query, engine, params[user_input])10.2 数据库文件权限确保SQLite数据库文件有适当的权限设置生产环境中不应将数据库文件放在web可访问目录设置适当的文件系统权限(如600)10.3 敏感数据保护处理敏感数据时考虑使用SQLite的加密扩展或改用支持透明加密的数据库系统不要在代码中硬编码数据库路径11. 替代方案比较11.1 直接使用sqlite3模块import sqlite3 import pandas as pd conn sqlite3.connect(store.db) df pd.read_sql_query(SELECT * FROM VARIABLES, conn) conn.close()优点无需额外依赖更轻量级缺点功能较少需要手动管理连接11.2 使用其他数据库适配器如psycopg2(PostgreSQL)、PyMySQL(MySQL)等# PostgreSQL示例 engine create_engine(postgresql://user:passwordlocalhost/dbname)优点适用于客户端-服务器型数据库支持更高级的数据库功能缺点需要设置数据库服务器更复杂的配置11.3 使用DuckDBDuckDB是另一种嵌入式分析型数据库import duckdb # 直接查询Pandas DataFrame df duckdb.query(SELECT * FROM df WHERE value 100).to_df()优点为分析工作负载优化与Pandas深度集成缺点较新的项目生态系统不如SQLite成熟12. 扩展应用12.1 与Jupyter Notebook集成在Jupyter中可以直接显示查询结果# 在Jupyter单元格中 query SELECT category, AVG(value) as avg_value FROM VARIABLES GROUP BY category ORDER BY avg_value DESC df pd.read_sql_query(query, engine) df.style.background_gradient()12.2 构建简单的数据仪表板结合Streamlit快速构建仪表板# dashboard.py import streamlit as st import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///store.db) st.title(销售数据仪表板) timeframe st.selectbox(选择时间段, [本月, 本季度, 本年]) query f SELECT product, SUM(amount) as total_sales FROM sales WHERE {{ 本月: strftime(%Y-%m, sale_date) strftime(%Y-%m, now), 本季度: strftime(%Y-%m, sale_date) strftime(%Y-%m, now, start of month, -2 month), 本年: strftime(%Y, sale_date) strftime(%Y, now) }}[timeframe] GROUP BY product ORDER BY total_sales DESC LIMIT 10 df pd.read_sql_query(query, engine) st.bar_chart(df.set_index(product))12.3 自动化数据报告使用Python自动化生成数据报告from datetime import datetime import pandas as pd from sqlalchemy import create_engine import matplotlib.pyplot as plt def generate_report(db_path, output_path): engine create_engine(fsqlite:///{db_path}) # 查询数据 sales_query SELECT strftime(%Y-%m, date) as month, SUM(amount) as total_sales FROM sales GROUP BY month ORDER BY month sales_df pd.read_sql_query(sales_query, engine) # 生成图表 plt.figure(figsize(10, 5)) sales_df.plot(xmonth, ytotal_sales, kindbar) plt.title(月度销售额趋势) plt.tight_layout() chart_path f{output_path}/sales_trend.png plt.savefig(chart_path) # 生成HTML报告 report_date datetime.now().strftime(%Y-%m-%d) html f html headtitle销售报告 - {report_date}/title/head body h1销售报告 - {report_date}/h1 img srcsales_trend.png alt销售趋势 {sales_df.to_html()} /body /html with open(f{output_path}/report.html, w) as f: f.write(html) # 使用示例 generate_report(sales.db, reports) 
长河君 · 2026/11/5 4:43:37 · 阅读 1.2k Pandas与SQLAlchemy连接SQLite数据库实战指南
1. 项目概述在数据分析工作中我们经常需要从各种数据源提取数据进行分析。SQLite作为轻量级的嵌入式数据库因其无需服务器、零配置的特点常被用作本地数据存储方案。而Pandas作为Python数据分析的核心库提供了强大的数据处理能力。本文将详细介绍如何使用Pandas与SQLAlchemy库连接并查询SQLite数据库实现数据的高效读取和处理。这个技术组合特别适合以下场景需要快速分析存储在SQLite中的中小规模数据希望将数据库查询结果直接转换为DataFrame进行后续分析开发需要数据持久化的本地应用程序进行数据ETL(提取、转换、加载)流程的原型开发2. 环境准备与工具选型2.1 为什么选择SQLAlchemy作为连接工具SQLAlchemy是Python中最流行的ORM(对象关系映射)工具之一它提供了统一的API来操作不同类型的数据库。相比直接使用SQLite3模块SQLAlchemy具有以下优势数据库无关性同一套代码可以适配多种数据库只需修改连接字符串连接池管理自动管理数据库连接提高性能SQL注入防护内置参数化查询机制与Pandas无缝集成可以直接将查询结果转换为DataFrame2.2 安装必要的Python包在开始之前我们需要确保环境中安装了以下包conda install pandas sqlalchemy或者使用pip安装pip install pandas sqlalchemy提示建议使用conda进行安装因为它能更好地处理依赖关系特别是当项目中还涉及其他科学计算包时。3. 连接SQLite数据库3.1 创建数据库连接引擎SQLAlchemy使用引擎(Engine)对象来管理数据库连接。创建引擎的代码如下from sqlalchemy import create_engine # Windows路径示例 db_path rD:\data\store.db engine create_engine(fsqlite:///{db_path}) # Mac/Linux路径示例 # db_path /home/user/data/store.db # engine create_engine(fsqlite:////{db_path})路径说明Windows系统需要使用原始字符串(r前缀)避免转义问题SQLite连接字符串格式为sqlite:///(三个斜杠)后接数据库文件路径Linux/Mac系统需要四个斜杠因为路径以/开头3.2 连接池配置优化默认情况下SQLAlchemy会维护一个连接池。对于SQLite这种轻量级数据库我们可以调整一些参数以获得更好的性能engine create_engine( fsqlite:///{db_path}, pool_size5, # 连接池大小 max_overflow10, # 允许超出pool_size的连接数 pool_timeout30, # 获取连接的超时时间(秒) pool_recycle3600 # 连接回收时间(秒) )注意SQLite是文件型数据库过多的并发连接可能会导致性能下降甚至文件锁定问题。对于大多数应用场景保持默认配置即可。4. 使用Pandas查询数据4.1 基本查询方法Pandas提供了read_sql_query函数来执行SQL查询并直接返回DataFrameimport pandas as pd # 简单查询 query SELECT * FROM VARIABLES df pd.read_sql_query(query, engine) # 查看前10行 print(df.head(10))4.2 高级查询技巧4.2.1 参数化查询为了防止SQL注入和提高性能应该使用参数化查询# 使用命名参数 query SELECT * FROM VARIABLES WHERE category :cat df pd.read_sql_query(query, engine, params{cat: sales}) # 使用位置参数 query SELECT * FROM VARIABLES WHERE value ? df pd.read_sql_query(query, engine, params[100])4.2.2 分页查询对于大型表应该使用分页查询避免内存问题page_size 1000 page_num 1 query f SELECT * FROM VARIABLES LIMIT {page_size} OFFSET {(page_num-1)*page_size} df pd.read_sql_query(query, engine)4.2.3 使用索引加速查询如果经常按某列查询可以先为该列创建索引# 创建索引(只需执行一次) with engine.connect() as conn: conn.execute(CREATE INDEX IF NOT EXISTS idx_category ON VARIABLES(category)) # 使用索引列查询 query SELECT * FROM VARIABLES WHERE category sales df pd.read_sql_query(query, engine)5. 数据类型处理5.1 SQLite与Pandas类型映射SQLite和Pandas的数据类型不完全一致需要注意以下映射关系SQLite类型Pandas类型说明INTEGERint64REALfloat64TEXTobject字符串BLOBobject二进制数据NULLNaN缺失值5.2 处理日期时间数据SQLite没有专门的日期类型通常以TEXT、INTEGER或REAL存储。Pandas可以自动转换常见格式# 假设date列存储为YYYY-MM-DD格式 query SELECT * FROM EVENTS df pd.read_sql_query(query, engine, parse_dates[date]) # 自定义日期格式 df[date] pd.to_datetime(df[date], format%Y/%m/%d)5.3 处理缺失值SQLite中的NULL值会被转换为Pandas的NaN。我们可以使用以下方法处理# 检查缺失值 print(df.isnull().sum()) # 填充缺失值 df.fillna({column1: 0, column2: unknown}, inplaceTrue) # 删除包含缺失值的行 df.dropna(subset[important_column], inplaceTrue)6. 性能优化技巧6.1 只查询需要的列避免使用SELECT *只查询必要的列query SELECT id, name, value FROM VARIABLES df pd.read_sql_query(query, engine)6.2 使用chunksize处理大数据对于大型表可以使用chunksize参数分批读取chunk_iter pd.read_sql_query(SELECT * FROM LARGE_TABLE, engine, chunksize10000) for chunk in chunk_iter: process(chunk) # 处理每个数据块6.3 使用索引加速查询确保查询条件使用索引列# 好的查询(使用索引列) query SELECT * FROM VARIABLES WHERE indexed_column ? # 不好的查询(全表扫描) query SELECT * FROM VARIABLES WHERE non_indexed_column ?6.4 批量插入数据使用Pandas的to_sql方法批量插入数据# 准备数据 new_data pd.DataFrame({ name: [A, B, C], value: [10, 20, 30] }) # 批量插入 new_data.to_sql(VARIABLES, engine, if_existsappend, indexFalse)参数说明if_existsappend追加数据而不是覆盖表indexFalse不写入DataFrame的索引列7. 常见问题与解决方案7.1 数据库锁定问题当多个进程同时访问SQLite数据库时可能会遇到database is locked错误。解决方案确保及时关闭连接with engine.connect() as conn: df pd.read_sql_query(query, conn)设置较长的超时时间engine create_engine(fsqlite:///{db_path}, connect_args{timeout: 30})使用WAL(Write-Ahead Logging)模式engine create_engine(fsqlite:///{db_path}, connect_args{timeout: 30, isolation_level: IMMEDIATE}) with engine.connect() as conn: conn.execute(PRAGMA journal_modeWAL)7.2 内存不足问题处理大型数据库时可能遇到内存不足。解决方法使用chunksize分批读取只查询需要的列在SQL端进行聚合过滤减少返回数据量7.3 数据类型转换问题当SQLite中的数据类型与Pandas预期不符时显式指定数据类型dtype { id: int64, name: str, value: float64 } df pd.read_sql_query(query, engine, dtypedtype)查询后转换类型df[column] df[column].astype(int64)7.4 路径问题在不同操作系统上部署时路径处理要小心from pathlib import Path db_path Path(data) / store.db # 跨平台路径构造 engine create_engine(fsqlite:///{db_path.absolute()})8. 实际应用案例8.1 数据分析工作流示例# 1. 连接数据库 engine create_engine(sqlite:///sales.db) # 2. 查询销售数据 query SELECT strftime(%Y-%m, sale_date) AS month, product_category, SUM(amount) AS total_sales FROM sales WHERE sale_date 2023-01-01 GROUP BY month, product_category ORDER BY month, total_sales DESC sales_df pd.read_sql_query(query, engine) # 3. 数据分析 top_categories sales_df.groupby(month).apply( lambda x: x.nlargest(3, total_sales) ) # 4. 可视化 import matplotlib.pyplot as plt for month, data in top_categories.groupby(month): data.plot.bar(xproduct_category, ytotal_sales, titlemonth) plt.show()8.2 数据清洗与转换# 从数据库读取原始数据 raw_df pd.read_sql_query(SELECT * FROM customer_data, engine) # 数据清洗 clean_df raw_df.copy() clean_df[email] clean_df[email].str.lower().str.strip() clean_df[phone] clean_df[phone].str.replace(r\D, , regexTrue) # 处理缺失值 clean_df[segment] clean_df[segment].fillna(unknown) # 保存清洗后的数据 clean_df.to_sql(clean_customer_data, engine, if_existsreplace, indexFalse)9. 高级技巧与最佳实践9.1 使用SQLAlchemy Core进行复杂查询对于复杂查询可以结合SQLAlchemy Core使用from sqlalchemy import Table, MetaData, select metadata MetaData() variables Table(VARIABLES, metadata, autoload_withengine) # 构建查询 query select( variables.c.id, variables.c.name ).where( variables.c.value 100 ).order_by( variables.c.name ) # 执行查询 df pd.read_sql_query(query, engine)9.2 事务管理确保数据一致性使用事务操作with engine.begin() as conn: # 查询数据 df pd.read_sql_query(SELECT * FROM ACCOUNTS WHERE balance 1000, conn) # 更新数据 conn.execute( UPDATE ACCOUNTS SET status reviewed WHERE balance 1000 )9.3 使用连接上下文管理器始终使用连接上下文管理器确保资源释放with engine.connect() as conn: df1 pd.read_sql_query(SELECT * FROM TABLE1, conn) df2 pd.read_sql_query(SELECT * FROM TABLE2, conn) # 可以在同一个连接上执行多个查询 result conn.execute(SELECT COUNT(*) FROM TABLE1) count result.scalar()9.4 性能监控与调优使用SQLAlchemy的事件系统监控查询性能from sqlalchemy import event import time event.listens_for(engine, before_cursor_execute) def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): context._query_start_time time.time() event.listens_for(engine, after_cursor_execute) def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): duration time.time() - context._query_start_time if duration 0.5: # 记录慢查询 print(fSlow query ({duration:.2f}s): {statement})10. 安全注意事项10.1 防止SQL注入永远不要直接拼接SQL查询字符串# 错误做法(危险!) user_input admin; DROP TABLE users;-- query fSELECT * FROM users WHERE username {user_input} # 正确做法(使用参数化查询) query SELECT * FROM users WHERE username ? df pd.read_sql_query(query, engine, params[user_input])10.2 数据库文件权限确保SQLite数据库文件有适当的权限设置生产环境中不应将数据库文件放在web可访问目录设置适当的文件系统权限(如600)10.3 敏感数据保护处理敏感数据时考虑使用SQLite的加密扩展或改用支持透明加密的数据库系统不要在代码中硬编码数据库路径11. 替代方案比较11.1 直接使用sqlite3模块import sqlite3 import pandas as pd conn sqlite3.connect(store.db) df pd.read_sql_query(SELECT * FROM VARIABLES, conn) conn.close()优点无需额外依赖更轻量级缺点功能较少需要手动管理连接11.2 使用其他数据库适配器如psycopg2(PostgreSQL)、PyMySQL(MySQL)等# PostgreSQL示例 engine create_engine(postgresql://user:passwordlocalhost/dbname)优点适用于客户端-服务器型数据库支持更高级的数据库功能缺点需要设置数据库服务器更复杂的配置11.3 使用DuckDBDuckDB是另一种嵌入式分析型数据库import duckdb # 直接查询Pandas DataFrame df duckdb.query(SELECT * FROM df WHERE value 100).to_df()优点为分析工作负载优化与Pandas深度集成缺点较新的项目生态系统不如SQLite成熟12. 扩展应用12.1 与Jupyter Notebook集成在Jupyter中可以直接显示查询结果# 在Jupyter单元格中 query SELECT category, AVG(value) as avg_value FROM VARIABLES GROUP BY category ORDER BY avg_value DESC df pd.read_sql_query(query, engine) df.style.background_gradient()12.2 构建简单的数据仪表板结合Streamlit快速构建仪表板# dashboard.py import streamlit as st import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///store.db) st.title(销售数据仪表板) timeframe st.selectbox(选择时间段, [本月, 本季度, 本年]) query f SELECT product, SUM(amount) as total_sales FROM sales WHERE {{ 本月: strftime(%Y-%m, sale_date) strftime(%Y-%m, now), 本季度: strftime(%Y-%m, sale_date) strftime(%Y-%m, now, start of month, -2 month), 本年: strftime(%Y, sale_date) strftime(%Y, now) }}[timeframe] GROUP BY product ORDER BY total_sales DESC LIMIT 10 df pd.read_sql_query(query, engine) st.bar_chart(df.set_index(product))12.3 自动化数据报告使用Python自动化生成数据报告from datetime import datetime import pandas as pd from sqlalchemy import create_engine import matplotlib.pyplot as plt def generate_report(db_path, output_path): engine create_engine(fsqlite:///{db_path}) # 查询数据 sales_query SELECT strftime(%Y-%m, date) as month, SUM(amount) as total_sales FROM sales GROUP BY month ORDER BY month sales_df pd.read_sql_query(sales_query, engine) # 生成图表 plt.figure(figsize(10, 5)) sales_df.plot(xmonth, ytotal_sales, kindbar) plt.title(月度销售额趋势) plt.tight_layout() chart_path f{output_path}/sales_trend.png plt.savefig(chart_path) # 生成HTML报告 report_date datetime.now().strftime(%Y-%m-%d) html f html headtitle销售报告 - {report_date}/title/head body h1销售报告 - {report_date}/h1 img srcsales_trend.png alt销售趋势 {sales_df.to_html()} /body /html with open(f{output_path}/report.html, w) as f: f.write(html) # 使用示例 generate_report(sales.db, reports)