当前位置：首页 > article >正文

PySpark实现Snowflake数据导出到Amazon Redshift

article 2025/3/5 6:12:25

编写AWS EMR上的高性能PySpark代码，实现用SQL从Snowflake上下载数据到S3里的parquet文件，并导入Redshift表。

步骤一：配置EMR集群

首先确保您已经在AWS EMR上正确地设置了包含适当权限的角色和安全组的集群。该角色应允许访问Snowflake数据库、S3桶以及Redshift集群。

步骤二：连接至Snowflake并提取数据

接下来，我们将使用snowflake-jdbc-driver.jar驱动程序来连接Snowflake数据库并将查询结果加载到Dataframe中。请根据实际情况替换占位符值（例如用户名、密码、账户名称等）。

from pyspark.sql import SparkSession

# 创建Spark会话对象
spark = SparkSession.builder.appName("LoadFromSnowflakeToS3").getOrCreate()

options_snowflake = {
    "sfURL": "<SNOWFLAKE_ACCOUNT>.snowflakecomputing.com",
    "sfUser": "<USERNAME>",
    "sfPassword": "<PASSWORD>",
    "sfDatabase": "<DATABASE_NAME>",
    "sfSchema": "<SCHEMA_NAME>",  
}

query = "(SELECT * FROM <TABLE>)"

data_df = spark.read.format("net.snowflake.spark.SNOWFLAKE_SOURCE_JDBC")\
                .option(**options_snowflake)\
                .option("dbtable", query).load()

注意事项: 您可能还需要添加其他选项，比如SSL证书路径或代理设置，具体取决于您的网络环境和安全性需求。

步骤三：保存为Parquet格式于S3

一旦我们有了所需的DataFrame (data_df) ，就可以轻松地将其序列化成高效的Parquet格式并存储在指定位置的S3 bucket里。

output_path = f"s3a://{bucket_name}/{folder}/"
data_df.write.mode('overwrite').format('parquet').save(output_path)

这里的 mode='overwrite' 表示每次运行都会覆盖已存在的同名目录；如果您希望保留旧版记录而不被新版本替代的话，请改为 'append' 。另外别忘了提供正确的IAM策略给EC2节点以便它们能够向目标Bucket写入数据。

步骤四：将Parquet文件载入Redshift

现在我们需要借助COPY命令将上述生成的Parquet文件批量复制进Redshift的一个表格里面。为此目的，通常我们会采用UNLOAD方法配合Manifest文件的方式来进行更复杂的场景管理，但在这里为了简化流程，我仅展示基本语法供参考：

方法A: 使用JDBC Driver 直接插入 Redshift (适用于小规模数据集)

这种方法并不推荐用于大规模作业，因为它依赖单个Executor线程逐条推送数据进去，效率低下且容易导致超时错误。

INSERT INTO redshift_table SELECT * FROM snowflake_view;

但是由于我们在之前已经完成了ETL过程并且得到了优化后的Parquet输出，所以建议采取第二种方案:

方法B: 利用 COPY 命令高效传输大容量 Parquet 文件

这是最佳实践方式，因为它是多线程化的，并能充分利用Redshift内部压缩机制提高吞吐率。

准备阶段:

创建外部分区表: 定义一个临时外表指向刚才存放在 S3 上的所有 .parqeut 文件片段。
授权 IAM 角色: 确保关联到 Redshift 的 EC2 实例拥有足够的权限去读取相应 Bucket 内容。

示例脚本：

CREATE EXTERNAL SCHEMA IF NOT EXISTS ext_schema_from_parquets
FROM DATA CATALOG DATABASE '<glue_database>'
REGION AS '<region>';

GRANT USAGE ON SCHEMA ext_schema_from_parquets TO GROUP your_group;

-- 如果没有Glue Catalog可用，则手动定义EXTERNAL TABLE如下所示：
/*
CREATE EXTERNAL TABLE schema.ext_table (
   column1 datatype,
   ...
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 's3://path/to/output/';
*/

-- 执行实际COPY操作
COPY target_redshift_table
FROM '{manifest_file}'
CREDENTIALS 'aws_access_key_id=<access-key-id>;aws_secret_access_key=<secret-access-key>' -- or better practice with iam role arn instead!
MANIFEST GZIP REGION '<region>';