从 Snowflake 迁移到 BigQuery 的教程

本文档提供了端到端示例和教程，介绍了设置从 Snowflake 到 BigQuery 的迁移流水线的不同方式。

Snowflake 迁移流水线示例

您可以使用三种不同的流程将数据从 Snowflake 迁移到 BigQuery：ELT、ETL 或使用合作伙伴工具。

提取、加载和转换

您可以通过以下两种方法设置提取、加载和转换 (ELT) 流程：

使用流水线从 Snowflake 提取数据并将数据加载到 BigQuery
使用其他 Google Cloud 产品从 Snowflake 提取数据。

使用流水线从 Snowflake 提取数据

如需从 Snowflake 提取数据并直接加载到 Cloud Storage，请使用 snowflake2bq 工具。

然后，您可以使用以下任一工具将数据从 Cloud Storage 加载到 BigQuery 中：

适用于 Cloud Storage 的 BigQuery Data Transfer Service 连接器
使用 bq 命令行工具的 LOAD 命令
BigQuery API 客户端库

用于从 Snowflake 提取数据的其他工具

您还可以使用以下工具从 Snowflake 提取数据：

Dataflow
- JDBC to BigQuery 模板
- SnowflakeIO 连接器
Cloud Data Fusion
- JDBC 驱动程序
Dataproc
- Apache Spark BigQuery 连接器
- 适用于 Apache Spark 的 Snowflake 连接器
- Hadoop BigQuery 连接器
- Snowflake 和 Sqoop 的 JDBC 驱动程序，用于将数据从 Snowflake 提取到 Cloud Storage 中：
  - 在 Dataproc 中使用 Apache Sqoop 移动数据

用于将数据加载到 BigQuery 的其他工具

您还可以使用以下工具将数据加载到 BigQuery：

Dataflow
Cloud Data Fusion
- 创建目标广告系列流水线
Dataproc
Dataprep by Trifacta
- 从 Cloud Storage 读取数据
- 写入 BigQuery

提取、转换和加载

如果您想在将数据加载到 BigQuery 之前转换数据，请考虑使用以下工具：

Dataflow
- 克隆 JDBC to BigQuery 模板代码并修改模板以添加 Apache Beam 转换。
Cloud Data Fusion
- 使用 CDAP 插件创建可重用的流水线并转换数据。
Dataproc
- 使用 Spark SQL 或是以任何受支持的 Spark 语言（Scala、Java、Python 或 R）编写的自定义代码来转换数据。

用于迁移的合作伙伴工具

有许多供应商专门提供 EDW 迁移空间。如需查看主要合作伙伴及其提供的解决方案的列表，请参阅 BigQuery 合作伙伴。

Snowflake 导出教程

以下教程展示了使用 COPY INTO <location> Snowflake 命令将数据从 Snowflake 导出到 BigQuery 的示例。如需查看包含代码示例的详细分步过程，请参阅Google Cloud 专业服务 Snowflake to BigQuery 工具。

准备导出

您可以按照以下步骤将 Snowflake 数据提取到 Cloud Storage 或 Amazon Simple Storage Service (Amazon S3) 存储桶中，从而为导出准备好 Snowflake 数据：

Cloud Storage

本教程会准备好 PARQUET 格式的文件。

使用 Snowflake SQL 语句创建一个命名文件格式规范。
```
create or replace file format NAMED_FILE_FORMAT
    type = 'PARQUET'
```
将 NAMED_FILE_FORMAT 替换为文件格式的名称。例如 my_parquet_unload_format。
使用 CREATE STORAGE INTEGRATION 命令创建一个集成。
```
create storage integration INTEGRATION_NAME
    type = external_stage
    storage_provider = gcs
    enabled = true
    storage_allowed_locations = ('BUCKET_NAME')
```
替换以下内容：
- INTEGRATION_NAME：存储集成的名称。例如 gcs_int
- BUCKET_NAME：Cloud Storage 存储桶的路径。例如 gcs://mybucket/extract/

使用 DESCRIBE INTEGRATION 命令检索 Snowflake 的 Cloud Storage 服务账号。

desc storage integration INTEGRATION_NAME;

输出类似于以下内容：

+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+
| property                    | property_type | property_value                                                              | property_default |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------|
| ENABLED                     | Boolean       | true                                                                        | false            |
| STORAGE_ALLOWED_LOCATIONS   | List          | gcs://mybucket1/path1/,gcs://mybucket2/path2/                               | []               |
| STORAGE_BLOCKED_LOCATIONS   | List          | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/   | []               |
| STORAGE_GCP_SERVICE_ACCOUNT | String        | service-account-id@iam.gserviceaccount.com                 |                  |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+

向列为 STORAGE_GCP_SERVICE_ACCOUNT 的服务账号授予对存储集成命令中指定的存储桶的读写权限。在此示例中，向 service-account-id@ 服务账号授予对 <var>UNLOAD_BUCKET</var> 存储桶的读写权限。
创建一个外部 Cloud Storage 暂存区，以引用您之前创建的集成。
```
create or replace stage STAGE_NAME
    url='UNLOAD_BUCKET'
    storage_integration = INTEGRATION_NAME
    file_format = NAMED_FILE_FORMAT;
```
替换以下内容：
- STAGE_NAME：Cloud Storage 暂存区对象的名称。例如 my_ext_unload_stage

Amazon S3

以下示例展示了如何将数据从 Snowflake 表移至 Amazon S3 存储桶：

在 Snowflake 中配置存储集成对象，以允许 Snowflake 写入外部 Cloud Storage 暂存区中引用的 Amazon S3 存储桶。

此步骤涉及配置对 Amazon S3 存储桶的访问权限、创建 Amazon Web Services (AWS) IAM 角色以及使用 CREATE STORAGE INTEGRATION 命令在 Snowflake 中创建存储集成：
```
create storage integration INTEGRATION_NAME
type = external_stage
storage_provider = s3
enabled = true
storage_aws_role_arn = 'arn:aws:iam::001234567890:role/myrole'
storage_allowed_locations = ('BUCKET_NAME')
```
替换以下内容：
- INTEGRATION_NAME：存储集成的名称。例如 s3_int
- BUCKET_NAME：要将文件加载到的 Amazon S3 存储桶的路径。例如 s3://unload/files/

使用 DESCRIBE INTEGRATION 命令检索 AWS IAM 用户。

desc integration INTEGRATION_NAME;

输出类似于以下内容：

+---------------------------+---------------+================================================================================+------------------+
| property                  | property_type | property_value                                                                 | property_default |
+---------------------------+---------------+================================================================================+------------------|
| ENABLED                   | Boolean       | true                                                                           | false            |
| STORAGE_ALLOWED_LOCATIONS | List          | s3://mybucket1/mypath1/,s3://mybucket2/mypath2/                                | []               |
| STORAGE_BLOCKED_LOCATIONS | List          | s3://mybucket1/mypath1/sensitivedata/,s3://mybucket2/mypath2/sensitivedata/    | []               |
| STORAGE_AWS_IAM_USER_ARN  | String        | arn:aws:iam::123456789001:user/abc1-b-self1234                                 |                  |
| STORAGE_AWS_ROLE_ARN      | String        | arn:aws:iam::001234567890:role/myrole                                          |                  |
| STORAGE_AWS_EXTERNAL_ID   | String        | MYACCOUNT_SFCRole=                                                   |                  |
+---------------------------+---------------+================================================================================+------------------+

创建对架构具有 CREATE STAGE 权限且对存储集成具有 USAGE 权限的角色：

    CREATE role ROLE_NAME;  
    GRANT CREATE STAGE ON SCHEMA public TO ROLE ROLE_NAME;
    GRANT USAGE ON INTEGRATION s3_int TO ROLE ROLE_NAME;

将 ROLE_NAME 替换为角色的名称。例如 myrole。

向 AWS IAM 用户授予 Amazon S3 存储桶的访问权限，并使用 CREATE STAGE 命令创建外部暂存区。

    USE SCHEMA mydb.public;

    create or replace stage STAGE_NAME
        url='BUCKET_NAME'
        storage_integration = INTEGRATION_NAMEt
        file_format = NAMED_FILE_FORMAT;

替换以下内容：

STAGE_NAME：Cloud Storage 暂存区对象的名称。例如 my_ext_unload_stage

导出 Snowflake 数据

准备好数据后，您可以将数据迁移到 Google Cloud。使用 COPY INTO 命令并指定外部暂存区对象 STAGE_NAME，以将数据从 Snowflake 数据库表复制到 Cloud Storage 或 Amazon S3 存储桶。

    copy into @STAGE_NAME/d1
    from TABLE_NAME;

将 TABLE_NAME 替换为您的 Snowflake 数据库表的名称。

执行此命令后，表数据会复制到与 Cloud Storage 或 Amazon S3 存储桶相关联的暂存区对象。文件包含 d1 前缀。

其他导出方法

如需使用 Azure Blob Storage 导出数据，请按照卸载到 Microsoft Azure 中详细介绍的步骤操作。然后，使用 Storage Transfer Service 将导出的文件转移到 Cloud Storage。