Amazon S3+AWS Glue+BladePipe快速实现数据实时入湖

目前Apache Iceberg已深度集成进 AWS 的数据湖生态系统，通过将 Iceberg 与 S3、Glue、Athena 等 AWS 服务结合，可以构建现代化的数据湖仓架构，实现更强的数据治理能力和分析性能。本文将介绍如何使用Amazon S3+AWS Glue+BladePipe五分钟快速构建数据管道，实现数据实时入湖。

一、BladePipe是什么

BladePipe是一款专业的端到端数据实时迁移同步工具，凭借其低延迟、高稳定性、自动化、可视化等优势，广泛应用于金融、医药、游戏、新能源汽车等领域。BladePip e支持 40+ 主流数据源，对于 Iceberg 这类非传统数据库交互的产品，BladePipe 也实现了数据迁移同步的自动化流程，包括结构定义转换、类型映射、约束清理、类型长度适配等工作，都可在 BladePipe 一站式完成。

BladePipe 支持 Iceberg 3 种 Catalog 和 2 种存储方式，搭配关系为：

AWS Glue + Amazon S3
Nessie + MinIO / Amazon S3
Rest + MinIO / Amazon S3

BladePipe目前支持从以下数据源同步数据至 Iceberg：

MySQL/MariaDB/AuroraMySQL
Oracle
PostgreSQL
SQL Server
Kafka

二、数据同步流程

三、搭建准备

1、准备存储桶

亚马逊云科技官网：点击直达（免费领取12个月Amazon S3服务）

通过上方链接访问注册亚马逊云科技，并在Amazon S3上创建一个存储桶。具体步骤参考《AWS S3使用教程》。

2、准备BladePipe

BladePipe提供私有部署（On-Premise）和 BYOC 两种部署模式。本文将以 BYOC 部署模式为例，介绍同步流程。

在 BYOC 部署模式下，登录 BladePipe 云平台，根据 Install Worker(Docker) 或 Install Worker(Binary) 的指引，下载安装 BladePipe Worker。

四、Amazon S3+AWS Glue+BladePipe快速搭建数据管道

本案例将展示从 MySQL（自建）到 Iceberg （AWS Glue + Amazon S3）的数据迁移同步。

1、添加数据源

添加 Iceberg 数据源时所要填写的信息如下（<>内按实际情况替换）。

网络地址：本例填写 AWS Glue 服务地址<aws_glue_region_code>.amazonaws.com；
版本：保持默认值即可；
描述：用于辨别实例用途；

额外参数配置：

httpsEnabled：打开开关，即设置为 true；
catalogName：设置一个意义明确的名字，如 glue_<biz_name>_catalog；
catalogType：设置为 GLUE；
catalogWarehouse：元数据和数据文件最终存放位置，如 s3://<biz_name>_iceberg；

catalogProps参考配置：

{
“io-impl”: “org.apache.iceberg.aws.s3.S3FileIO”,
“s3.endpoint”: “https://s3.<aws_s3_region_code>.amazonaws.com”,
“s3.access-key-id”: “<aws_s3_iam_user_access_key>”,
“s3.secret-access-key”: “<aws_s3_iam_user_secret_key>”,
“s3.path-style-access”: “true”,
“client.region”: “<aws_s3_region>”,
“client.credentials-provider.glue.access-key-id”: “<aws_glue_iam_user_access_key>”,
“client.credentials-provider.glue.secret-access-key”: “<aws_glue_iam_user_secret_key>”,
“client.credentials-provider”: “com.amazonaws.glue.catalog.credentials.GlueAwsCredentialsProvider”
}

2、创建任务

在 BladePipe 平台，点击同步任务 > 创建任务。选择源（MySQL）和目标（Iceberg）实例，并分别点击测试连接。其中 Iceberg 数据源结构迁移属性配置推荐如下：

{
“format-version”: “2”,
“parquet.compression”: “snappy”,
“iceberg.write.format”: “parquet”,
“write.metadata.delete-after-commit.enabled”: “true”,
“write.metadata.previous-versions-max”: “3”,
“write.update.mode”: “merge-on-read”,
“write.delete.mode”: “merge-on-read”,
“write.merge.mode”: “merge-on-read”,
“write.distribution-mode”: “hash”,
“write.object-storage.enabled”: “true”,
“write.spark.accept-any-schema”: “true”
}

如遇到测试连接长时间不返回，可以刷新页面重新选择。数据库连接信息错误或网络不通都可能造成该现象。