如何在Jupyter Lab中并行管理多数据库连接并批量拉取数据

我需要连接PostgreSQL、Snowflake和MySQL多个数据源，从各库中提取指定表用于模型构建。已知可通过SQLAlchemy或JupySQL单独操作，能否通过代码复用，一次性初始化各连接，循环传入查询语句批量拉取表数据并本地合并？

完全可以实现，下面分两种常用方案给出具体实现代码：

方案一：基于JupySQL实现批量连接与数据拉取 #

JupySQL结合Python逻辑可快速实现批量操作，适合Notebook环境快速验证：

1. 安装依赖 #

先安装JupySQL及各数据库驱动包：

pip install jupysql psycopg2-binary mysql-connector-python snowflake-sqlalchemy --quiet

2. 批量连接与数据拉取 #

%load_ext sql
import pandas as pd

# 定义所有数据源的连接配置及对应查询语句
data_sources = [
    {
        "conn_url": "postgresql://username:password@host:port/database",
        "query": "SELECT * FROM pg_target_table"
    },
    {
        "conn_url": "mysql+mysqlconnector://username:password@host:port/database",
        "query": "SELECT * FROM mysql_target_table"
    },
    {
        "conn_url": "snowflake://username:password@account_identifier/database/schema?warehouse=warehouse_name",
        "query": "SELECT * FROM snowflake_target_table"
    }
]

# 初始化空列表存储各数据源的数据
data_frames = []

# 循环处理每个数据源
for source in data_sources:
    # 切换当前连接
    %sql {source["conn_url"]}
    # 执行查询并转换为DataFrame
    df = %sql {source["query"]}
    data_frames.append(df.DataFrame())

# 合并所有数据（按需选择concat/merge，此处以横向拼接为例）
merged_data = pd.concat(data_frames, ignore_index=True)

方案二：基于SQLAlchemy实现（适合脚本化生产场景） #

SQLAlchemy是纯Python ORM工具，无魔法命令，更适合生产环境脚本：

1. 安装依赖 #

pip install sqlalchemy psycopg2-binary mysql-connector-python snowflake-sqlalchemy pandas --quiet

2. 批量连接与数据拉取 #

from sqlalchemy import create_engine
import pandas as pd

# 定义数据源配置
data_sources = [
    {
        "engine_url": "postgresql://username:password@host:port/database",
        "query": "SELECT * FROM pg_target_table"
    },
    {
        "engine_url": "mysql+mysqlconnector://username:password@host:port/database",
        "query": "SELECT * FROM mysql_target_table"
    },
    {
        "engine_url": "snowflake://username:password@account_identifier/database/schema?warehouse=warehouse_name",
        "query": "SELECT * FROM snowflake_target_table"
    }
]

data_frames = []

for source in data_sources:
    # 创建数据库引擎并建立连接
    engine = create_engine(source["engine_url"])
    with engine.connect() as conn:
        # 执行查询并读取为DataFrame
        df = pd.read_sql(source["query"], conn)
        data_frames.append(df)
    # 关闭引擎释放资源
    engine.dispose()

# 合并数据（按需调整合并逻辑）
merged_data = pd.concat(data_frames, ignore_index=True)

注意事项 #

• 替换配置中的username、password、host等参数为实际环境值
• 若各表结构差异较大，需先做字段对齐，再选择合适的合并方式（concat或merge）
• Snowflake的连接URL需根据实际账户信息调整，确保包含仓库、数据库、Schema等必填参数

内容的提问来源于stack exchange，提问作者Ido Michael