Python怎样操作Amazon Redshift？sqlalchemy连接

SQLAlchemy

psycopg2

pip install sqlalchemy psycopg2-binary

from sqlalchemy import create_engine, text
import os

# 生产环境强烈不建议硬编码凭证，这里仅为演示
# 更好的实践是使用环境变量、配置文件或AWS Secrets Manager
REDSHIFT_USER = os.environ.get("REDSHIFT_USER", "your_redshift_user")
REDSHIFT_PASSWORD = os.environ.get("REDSHIFT_PASSWORD", "your_redshift_password")
REDSHIFT_HOST = os.environ.get("REDSHIFT_HOST", "your-redshift-cluster.xxxx.region.redshift.amazonaws.com")
REDSHIFT_PORT = os.environ.get("REDSHIFT_PORT", "5439")
REDSHIFT_DB = os.environ.get("REDSHIFT_DB", "your_database_name")

# 构建连接字符串
# 注意：Redshift通常建议开启SSL，可以加上 sslmode=require
# 如果你遇到了连接问题，可以尝试加上 sslmode=require 或 sslmode=prefer
# connection_string = f"postgresql+psycopg2://{REDSHIFT_USER}:{REDSHIFT_PASSWORD}@{REDSHIFT_HOST}:{REDSHIFT_PORT}/{REDSHIFT_DB}?sslmode=require"
connection_string = f"postgresql+psycopg2://{REDSHIFT_USER}:{REDSHIFT_PASSWORD}@{REDSHIFT_HOST}:{REDSHIFT_PORT}/{REDSHIFT_DB}"

try:
    # 创建引擎
    # pool_recycle 参数可以帮助处理长时间不活动的连接断开问题
    engine = create_engine(connection_string, pool_recycle=3600)

    # 使用 with 语句管理连接，确保连接在使用完毕后被正确关闭或放回连接池
    with engine.connect() as connection:
        # 执行一个简单的查询
        print("尝试连接Redshift并执行查询...")
        result = connection.execute(text("SELECT current_user, GETDATE() AS current_redshift_time;"))

        # 获取结果
        for row in result:
            print(f"当前Redshift用户: {row.current_user}, Redshift时间: {row.current_redshift_time}")

        # 示例：执行一个数据操作（比如创建一个临时表并插入数据）
        # connection.execute(text("DROP TABLE IF EXISTS my_test_table;"))
        # connection.execute(text("CREATE TABLE my_test_table (id INT, name VARCHAR(100));"))
        # connection.execute(text("INSERT INTO my_test_table (id, name) VALUES (1, 'Alice'), (2, 'Bob');"))
        # connection.commit() # 对于DML操作，需要提交事务

        # 再次查询新创建的表
        # print("\n查询 my_test_table:")
        # new_result = connection.execute(text("SELECT * FROM my_test_table;"))
        # for row in new_result:
        #     print(f"ID: {row.id}, Name: {row.name}")

    print("\nRedshift操作成功完成。")

except Exception as e:
    print(f"连接或查询Redshift时发生错误: {e}")

STV_QUERY_SUMMARY

SVL_QUERY_SUMMARY

create_engine

engine.connect()

create_engine

pool_size

max_overflow

pool_recycle

pd.read_sql()

LIMIT

OFFSET

cursor

UNLOAD

REDSHIFT_USER

REDSHIFT_PASSWORD

os.environ.get()

.env

config.ini

python-dotenv

.env

import boto3
import json

# 示例：从Secrets Manager获取凭证
def get_secret():
    secret_name = "your/redshift/secret/name" # 在Secrets Manager中定义的密钥名称
    region_name = "your-aws-region" # 你的AWS区域

    # 创建Secrets Manager客户端
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except Exception as e:
        # 错误处理，比如密钥不存在、权限不足等
        raise e
    else:
        # 密钥值是字符串，可能是JSON格式
        if 'SecretString' in get_secret_value_response:
            secret = get_secret_value_response['SecretString']
            return json.loads(secret)
        else:
            # 对于二进制密钥，需要进行Base64解码
            # 这里我们假设是字符串密钥
            raise ValueError("Secret is not a string type.")

# 使用示例
# if __name__ == "__main__":
#     try:
#         secrets = get_secret()
#         REDSHIFT_USER = secrets['username']
#         REDSHIFT_PASSWORD = secrets['password']
#         REDSHIFT_HOST = secrets['host']
#         REDSHIFT_PORT = secrets['port']
#         REDSHIFT_DB = secrets['dbname']
#         # 剩下的连接逻辑就和上面一样了
#         print("成功从Secrets Manager获取凭证！")
#     except Exception as e:
#         print(f"获取凭证失败: {e}")

redshift_connector

psycopg2

aws_iam_authenticator

INSERT

COPY

COPY

COPY

# 示例：从S3加载数据到Redshift
# data_file_path_on_s3 = "s3://your-bucket/your-data-folder/data.csv"
# aws_access_key_id = "..." # 最好使用IAM角色，而不是直接在这里写AK/SK
# aws_secret_access_key = "..."
#
# copy_sql = f"""
#     COPY your_table_name
#     FROM '{data_file_path_on_s3}'
#     IAM_ROLE 'arn:aws:iam::123456789012:role/YourRedshiftCopyRole'
#     FORMAT AS CSV
#     DELIMITER ','
#     IGNOREHEADER 1
#     REGION 'your-aws-region';
# """
# 或者
# copy_sql = f"""
#     COPY your_table_name
#     FROM '{data_file_path_on_s3}'
#     ACCESS_KEY_ID '{aws_access_key_id}'
#     SECRET_ACCESS_KEY '{aws_secret_access_key}'
#     FORMAT AS CSV
#     DELIMITER ','
#     IGNOREHEADER 1;
# """
#
# with engine.connect() as connection:
#     connection.execute(text(copy_sql))
#     connection.commit()

import pandas as pd

# with engine.connect() as connection:
#     # 使用 Pandas 的 read_sql_query，它会自动处理大部分细节
#     df = pd.read_sql_query("SELECT * FROM your_large_table;", connection)
#     print(f"读取到 {len(df)} 行数据。")

pd.read_sql_query

INSERT

UPDATE

DELETE

connection.commit()

connection.rollback()

with engine.connect() as connection: