はじめに
本記事はS3 Tables × Icebergシリーズの第3弾（完結編)です。
| # | テーマ | 記事 |
|---|---|---|
| 1 | S3 Tables + Apache Icebergのテーブルを構築、Athenaからのクエリ、Glue JobでのETLを実施 | 前々回 |
| 2 | EventBridge + Step Functionsで自動ETLパイプラインを構築 | 前回 |
| 3 | Redshift Serverlessでレイクハウスクエリ + Glue Data Qualityで品質検査 | 本記事 |
第1弾でデータの入れ物を作り、第2弾でデータを自動で流す仕組みを構築しました。
しかし、本番運用ではまだ2つの課題が残っています。
課題1: クエリ性能と分析の柔軟性
Athenaは手軽ですが、大量データの結合やウィンドウ関数を多用する複雑な分析には限界があります。Redshift Serverlessから同じIcebergテーブルを直接クエリできれば、データのコピーなしで高速な分析基盤が手に入ります。
課題2: データ品質の担保
現在のパイプラインはファイルが来たら投入するだけです。中身が不正なCSVや型が壊れたデータもそのままIcebergテーブルに入ってしまいます。投入前に品質チェックを入れ、不合格データはブロックする仕組みが必要です。
本記事では、この2つをTerraformで構築し、レイクハウスを完成させます。
この記事でやること
- Redshift ServerlessをTerraformでプロビジョニング
- RedshiftからS3 TablesのIcebergテーブルを直接クエリ（External Database経由）
- RedshiftからIcebergテーブルへの書き込み（INSERT / CTAS）
- Glue Job内にGlue Data Quality（DQDL）の品質チェックを組み込み
- 品質チェック不合格時の自動ブロック + SNS通知
- 全リソースをTerraformでIaC化
用語・技術をざっくり理解する
記事全体で登場する用語を、先にまとめて解説します。
Redshift 関連
| 用語 | 解説 |
|---|---|
| Amazon Redshift | AWSのデータウェアハウス。大量データに対する複雑なSQL分析を高速に処理するための専用エンジン。列指向ストレージ・大規模並列処理（MPP）が特徴 |
| Redshift Serverless | サーバー管理不要のRedshift。ノード数やクラスタ構成を気にせず使える。使った分だけ課金（RPU 単位）。 |
| RPU | Redshiftの処理能力の単位。Redshift Processing Unit。ServerlessではベースキャパシティをRPUで指定する（最小 4 RPU、1 RPU = 16 GB メモリ）。 |
| Namespace | テーブル・スキーマ・ユーザー・暗号化キー等を管理する論理グループ |
| Workgroup | RPU 数・VPC・セキュリティグループ等を管理する論理グループ。Namespaceと1:1で紐づく。 |
| External Database | 外部カタログへの参照。Redshiftの外にあるGlue Data CatalogのデータベースをRedshiftから直接参照する仕組み |
データ品質関連
| 用語 | 解説 |
|---|---|
| AWS Glue Data Quality | データ品質チェックサービス。DQDLでルールを定義し、データが条件を満たすか自動評価する。Glue ETLジョブ内で実行可能。 |
| DQDL | Data Quality Definition Language。AWS 独自のデータ品質ルール定義言語。IsComplete, IsUnique, ColumnValues 等のルールを記述する |
| 品質ゲート | 品質チェックに不合格のデータを下流に流さない仕組み |
| 品質スコア | ルール通過率。定義したルールのうち、何%がパスしたかの指標（例: 8/10 ルール通過 = 80%） |
前提条件
| 項目 | 要件 |
|---|---|
| 第1弾記事 | S3 Tables + Icebergテーブルが構築済み |
| 第2弾記事 | EventBridge + Step Functionsパイプラインが構築済み |
| リージョン | ap-northeast-1（東京） |
| Terraform | >= 1.5.0 |
| AWS Provider | >= 5.82.0 |
アーキテクチャ（完成版）
Glue JobにGlue Data Qualityによるデータ品質チェックを組み込みます。合格ならS3TablesのIcebergテーブルへINSERTを行い、不合格ならS3の隔離フォルダに対象の生データを移行します。また、Redshift ServerlessでのIcebergテーブルに対するクエリを実行します。
Terraformディレクトリ構成
前回記事のディレクトリ構成に今記事用のリソースを追加しました。
s3-tables-iceberg-lab/
├── main.tf ← モジュールの呼び出し
├── providers.tf ← プロバイダー情報
├── variables.tf ← 変数
└── modules/ ← 各モジュールを格納
├── s3_tables/ ← S3テーブル関連
│ ├── main.tf
│ ├── variables.tf
│ └── outputs.tf
├── s3_landing/ ← 生データ格納S3関連
│ ├── main.tf
│ ├── variables.tf
│ └── outputs.tf
├── iam/ ← IAM権限関連
│ ├── main.tf
│ ├── variables.tf
│ └── outputs.tf
├── glue/ ← glue関連
│ ├── main.tf
│ ├── variables.tf
│ ├── outputs.tf
│ └── scripts/
├── step_functions/ ← step functions関連
│ ├── main.tf
│ ├── variables.tf
│ └── outputs.tf
└── redshift/ ← redshift関連
├── main.tf
└── variables.tf
Part A: Redshift Serverless × Iceberg
Step 1：Redshift ServerlessをTerraformでプロビジョニング
modules/redshift/main.tf
Redshift Serverlessの各種設定を定義します。
# Namespace（データベース・ユーザーの入れ物）
resource "aws_redshiftserverless_namespace" "main" {
namespace_name = "${var.project_name}-ns"
db_name = "dev"
admin_username = "admin"
admin_user_password = var.redshift_admin_password
# RedshiftがS3/Glue/Lake FormationにアクセスするためのIAMロール
iam_roles = [var.redshift_iam_role_arn]
lifecycle {
ignore_changes = [admin_user_password] # 初回作成後はパスワードの差分を無視する
}
tags = {
Project = var.project_name
}
}
# Workgroup（計算リソースの入れ物）
resource "aws_redshiftserverless_workgroup" "main" {
namespace_name = aws_redshiftserverless_namespace.main.namespace_name
workgroup_name = "${var.project_name}-wg"
# ベースキャパシティ: 4RPU（最小値、検証用）
# 制限: 最大ストレージ32TB、テーブルあたり最大100カラム、メモリ64GB
base_capacity = 4
# パブリックアクセス（検証用。本番ではfalseにしてVPC内からのみ）
publicly_accessible = true
tags = {
Project = var.project_name
}
}
# 使用量制限（コスト暴走防止）
resource "aws_redshiftserverless_usage_limit" "daily" {
resource_arn = aws_redshiftserverless_workgroup.main.arn
usage_type = "serverless-compute"
amount = 8 # 1日あたり最大8RPU時間
period = "daily"
breach_action = "deactivate" # 上限到達でクエリ停止
}
modules/redshift/variables.tf
redshiftモジュールのmain.tfで使用する変数を定義します。
variable "project_name" {
description = "プロジェクト名"
type = string
default = "iceberg-lab"
}
variable "redshift_admin_password" {
description = "Redshift Serverless管理者パスワード"
type = string
sensitive = true
}
variable "redshift_iam_role_arn" {
description = "redshift用IAMロールARN"
type = string
}
modules/iam/main.tf
今記事で必要な各リソースが処理を実行するためのIAM権限を定義します。
#glue用IAMロール
resource "aws_iam_role" "glue" {
name = "${var.project_name}-glue-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = { Service = "glue.amazonaws.com" }
}]
})
}
#glue用IAMポリシーアタッチ
resource "aws_iam_role_policy_attachment" "glue_service" {
role = aws_iam_role.glue.name
policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}
#glue用IAMポリシー
resource "aws_iam_role_policy" "glue_data_access" {
name = "${var.project_name}-glue-data-access"
role = aws_iam_role.glue.id
# 生データランディング用バケット、S3Tables用バケット、Lakeformation、Glueに対する権限
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Sid = "S3LandingAccess"
Effect = "Allow"
Action = ["s3:GetObject", "s3:ListBucket"]
Resource = [
var.landing_bucket_arn,
"${var.landing_bucket_arn}/*",
var.scripts_bucket_arn,
"${var.scripts_bucket_arn}/*",
]
},
{
Sid = "S3TablesAccess"
Effect = "Allow"
Action = ["s3tables:*"]
Resource = [
var.s3tables_bucket_arn,
"${var.s3tables_bucket_arn}/*"
]
},
{
Sid = "GlueAndLakeFormation"
Effect = "Allow"
Action = ["glue:*", "lakeformation:GetDataAccess"]
Resource = ["*"]
}
]
})
}
# EventBridge用ロール
# EventBridgeがStep Functionsを起動するために必要
resource "aws_iam_role" "eventbridge" {
name = "${var.project_name}-eventbridge-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "events.amazonaws.com"
}
}
]
})
}
resource "aws_iam_role_policy" "eventbridge_start_sfn" {
name = "start-step-functions"
role = aws_iam_role.eventbridge.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = "states:StartExecution"
Resource = var.etl_pipeline_statemachine_arn
}
]
})
}
# Step Functions用ロール
# Step FunctionsがGlue Job を起動し、SNSに通知するために必要
resource "aws_iam_role" "step_functions" {
name = "${var.project_name}-sfn-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "states.amazonaws.com"
}
}
]
})
}
resource "aws_iam_role_policy" "sfn_glue_sns" {
name = "glue-and-sns"
role = aws_iam_role.step_functions.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
# Glue Jobの起動・ステータス確認
Effect = "Allow"
Action = [
"glue:StartJobRun",
"glue:GetJobRun",
"glue:GetJobRuns",
"glue:BatchStopJobRun"
]
Resource = var.csv_to_iceberg_glue_job_arn
},
{
# SNSへの通知
Effect = "Allow"
Action = "sns:Publish"
Resource = var.pipeline_notification_sns_topic_arn
}
]
})
}
################################以下今回追加箇所#####################################
# Redshift用ロール
resource "aws_iam_role" "redshift" {
name = "${var.project_name}-redshift-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "redshift.amazonaws.com"
}
}
]
})
}
resource "aws_iam_role_policy" "redshift_lakehouse" {
name = "lakehouse-access"
role = aws_iam_role.redshift.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
# Glue Data Catalogへのアクセス
Effect = "Allow"
Action = [
"glue:GetDatabase",
"glue:GetDatabases",
"glue:GetTable",
"glue:GetTables",
"glue:GetPartitions",
"glue:GetCatalog",
"glue:GetCatalogs",
"glue:CreateTable",
"glue:UpdateTable"
]
Resource = "*"
},
{
# S3 Tables（Iceberg テーブル）へのアクセス
Effect = "Allow"
Action = [
"s3tables:*"
]
Resource = "*"
},
{
# S3データ読み書き
Effect = "Allow"
Action = [
"s3:GetObject",
"s3:PutObject",
"s3:ListBucket",
"s3:GetBucketLocation"
]
Resource = "*"
},
{
# Lake Formation権限
Effect = "Allow"
Action = [
"lakeformation:GetDataAccess"
]
Resource = "*"
}
]
})
}
modules/iam/output.tf
他モジュールで使用する出力を定義します。
output "glue_iamrole_arn" {
value = aws_iam_role.glue.arn
}
output "eventbridge_iamrole_arn" {
value = aws_iam_role.eventbridge.arn
}
output "step_functions_iam_role_arn" {
value = aws_iam_role.step_functions.arn
}
################################以下今回追加箇所#####################################
output "redshift_iam_role_arn" {
value = aws_iam_role.redshift.arn
}
main.tf
モジュールの呼び出し側となるtfファイルを定義します。
# S3テーブル関連
module "s3_tables" {
source = "./modules/s3_tables/"
project_name = var.project_name
}
# 生データ格納S3関連
module "s3_landing" {
source = "./modules/s3_landing/"
project_name = var.project_name
etl_pipeline_statemachine_arn = module.step_functions.etl_pipeline_statemachine_arn
eventbridge_iamrole_arn = module.iam.eventbridge_iamrole_arn
}
# IAM関連
module "iam" {
source = "./modules/iam/"
project_name = var.project_name
landing_bucket_arn = module.s3_landing.landing_bucket_arn
s3tables_bucket_arn = module.s3_tables.s3tables_bucket_arn
scripts_bucket_arn = module.glue.scripts_bucket_arn
etl_pipeline_statemachine_arn = module.step_functions.etl_pipeline_statemachine_arn
csv_to_iceberg_glue_job_arn = module.glue.csv_to_iceberg_glue_job_arn
pipeline_notification_sns_topic_arn = module.step_functions.pipeline_notification_sns_topic_arn
}
# Glue関連
module "glue" {
source = "./modules/glue/"
project_name = var.project_name
s3tables_bucket_name = module.s3_tables.s3tables_bucket_name
glue_iamrole_arn = module.iam.glue_iamrole_arn
}
# StepFunctions関連
module "step_functions" {
source = "./modules/step_functions/"
project_name = var.project_name
step_functions_iam_role_arn = module.iam.step_functions_iam_role_arn
csv_to_iceberg_glue_job_name = module.glue.csv_to_iceberg_glue_job_name
s3tables_bucket_name = module.s3_tables.s3tables_bucket_name
notification_email = var.notification_email
}
################################以下今回追加箇所#####################################
# Redshift関連
module "redshift" {
source = "./modules/redshift/"
project_name = var.project_name
redshift_admin_password = var.redshift_admin_password
redshift_iam_role_arn = module.iam.redshift_iam_role_arn
}
デプロイ
まずは新しく作成したモジュールを読み込むためにterraform initを実行します。
C:\s3-tables-iceberg-lab > terraform init
Initializing the backend...
Initializing modules...
- redshift in modules\redshift
Initializing provider plugins...
- Reusing previous version of hashicorp/aws from the dependency lock file
- Using previously-installed hashicorp/aws v6.33.0
Terraform has been successfully initialized!
You may now begin working with Terraform. Try running "terraform plan" to see
any changes that are required for your infrastructure. All Terraform commands
should now work.
If you ever set or change modules or backend configuration for Terraform,
rerun this command to reinitialize your working directory. If you forget, other
commands will detect it and remind you to do so if necessary.
次にterraform planを実行します。実行前にTF_VAR_変数名という環境変数を設定します。これにより、モジュール呼び出しのmain.tfにて使用している変数redshift_admin_passwordに機微情報であるパスワードをコードに記載せずに値を設定できます。
C:\s3-tables-iceberg-lab > set TF_VAR_redshift_admin_password=xxx
C:\s3-tables-iceberg-lab > terraform plan
module.s3_tables.data.aws_caller_identity.current: Reading...
module.s3_landing.data.aws_caller_identity.current: Reading...
module.glue.data.aws_caller_identity.current: Reading...
module.step_functions.data.aws_caller_identity.current: Reading...
module.iam.aws_iam_role.eventbridge: Refreshing state... [id=iceberg-lab-eventbridge-role]
module.iam.aws_iam_role.step_functions: Refreshing state... [id=iceberg-lab-sfn-role]
module.step_functions.aws_sns_topic.pipeline_notification: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccoundId}:iceberg-lab-pipeline-notification]
module.iam.aws_iam_role.glue: Refreshing state... [id=iceberg-lab-glue-role]
module.s3_tables.aws_s3tables_table_bucket.main: Refreshing state... [name=iceberg-lab]
module.step_functions.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.s3_tables.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.glue.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.s3_landing.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.glue.aws_s3_bucket.glue_scripts: Refreshing state... [id=iceberg-lab-glue-scripts-{AccoundId}]
module.s3_landing.aws_s3_bucket.landing: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.step_functions.aws_sns_topic_subscription.email: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccoundId}:iceberg-lab-pipeline-notification:e0e63b7a-82f3-44ea-8a59-ba38e5f2af88]
module.s3_tables.aws_s3tables_namespace.analytics: Refreshing state...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Reading...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Read complete after 0s [id=3931547792]
module.s3_tables.aws_s3tables_table_bucket_policy.main: Refreshing state...
module.s3_tables.aws_s3tables_table.orders: Refreshing state... [name=orders]
module.s3_tables.aws_s3tables_table.customers: Refreshing state... [name=customers]
module.glue.aws_s3_object.glue_script: Refreshing state... [id=iceberg-lab-glue-scripts-{AccoundId}/scripts/glue_csv_to_iceberg.py]
module.s3_landing.aws_s3_bucket_notification.landing_eventbridge: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.s3_landing.aws_cloudwatch_event_rule.s3_csv_arrived: Refreshing state... [id=iceberg-lab-csv-arrived]
module.s3_landing.aws_s3_bucket_versioning.landing: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.iam.aws_iam_role_policy_attachment.glue_service: Refreshing state... [id=iceberg-lab-glue-role/arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole]
module.iam.aws_iam_role_policy.glue_data_access: Refreshing state... [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
module.glue.aws_glue_job.csv_to_iceberg: Refreshing state... [id=iceberg-lab-csv-to-iceberg]
module.iam.aws_iam_role_policy.sfn_glue_sns: Refreshing state... [id=iceberg-lab-sfn-role:glue-and-sns]
module.step_functions.aws_sfn_state_machine.etl_pipeline: Refreshing state... [id=arn:aws:states:ap-northeast-1:{AccoundId}:stateMachine:iceberg-lab-etl-pipeline]
module.iam.aws_iam_role_policy.eventbridge_start_sfn: Refreshing state... [id=iceberg-lab-eventbridge-role:start-step-functions]
module.s3_landing.aws_cloudwatch_event_target.step_functions: Refreshing state... [id=iceberg-lab-csv-arrived-terraform-20260225153517622700000001]
Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the
following symbols:
+ create
Terraform will perform the following actions:
# module.iam.aws_iam_role.redshift will be created
+ resource "aws_iam_role" "redshift" {
+ arn = (known after apply)
+ assume_role_policy = jsonencode(
{
+ Statement = [
+ {
+ Action = "sts:AssumeRole"
+ Effect = "Allow"
+ Principal = {
+ Service = "redshift.amazonaws.com"
}
},
]
+ Version = "2012-10-17"
}
)
+ create_date = (known after apply)
+ force_detach_policies = false
+ id = (known after apply)
+ managed_policy_arns = (known after apply)
+ max_session_duration = 3600
+ name = "iceberg-lab-redshift-role"
+ name_prefix = (known after apply)
+ path = "/"
+ tags_all = (known after apply)
+ unique_id = (known after apply)
+ inline_policy (known after apply)
}
# module.iam.aws_iam_role_policy.redshift_lakehouse will be created
+ resource "aws_iam_role_policy" "redshift_lakehouse" {
+ id = (known after apply)
+ name = "lakehouse-access"
+ name_prefix = (known after apply)
+ policy = jsonencode(
{
+ Statement = [
+ {
+ Action = [
+ "glue:GetDatabase",
+ "glue:GetDatabases",
+ "glue:GetTable",
+ "glue:GetTables",
+ "glue:GetPartitions",
+ "glue:GetCatalog",
+ "glue:GetCatalogs",
+ "glue:CreateTable",
+ "glue:UpdateTable",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "s3tables:*",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:ListBucket",
+ "s3:GetBucketLocation",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "lakeformation:GetDataAccess",
]
+ Effect = "Allow"
+ Resource = "*"
},
]
+ Version = "2012-10-17"
}
)
+ role = (known after apply)
}
# module.redshift.aws_redshiftserverless_namespace.main will be created
+ resource "aws_redshiftserverless_namespace" "main" {
+ admin_password_secret_arn = (known after apply)
+ admin_password_secret_kms_key_id = (known after apply)
+ admin_user_password = (sensitive value)
+ admin_user_password_wo = (write-only attribute)
+ admin_username = (sensitive value)
+ arn = (known after apply)
+ db_name = "dev"
+ iam_roles = (known after apply)
+ id = (known after apply)
+ kms_key_id = (known after apply)
+ namespace_id = (known after apply)
+ namespace_name = "iceberg-lab-ns"
+ region = "ap-northeast-1"
+ tags = {
+ "Project" = "iceberg-lab"
}
+ tags_all = {
+ "Project" = "iceberg-lab"
}
}
# module.redshift.aws_redshiftserverless_usage_limit.daily will be created
+ resource "aws_redshiftserverless_usage_limit" "daily" {
+ amount = 8
+ arn = (known after apply)
+ breach_action = "deactivate"
+ id = (known after apply)
+ period = "daily"
+ region = "ap-northeast-1"
+ resource_arn = (known after apply)
+ usage_type = "serverless-compute"
}
# module.redshift.aws_redshiftserverless_workgroup.main will be created
+ resource "aws_redshiftserverless_workgroup" "main" {
+ arn = (known after apply)
+ base_capacity = 4
+ endpoint = (known after apply)
+ id = (known after apply)
+ namespace_name = "iceberg-lab-ns"
+ port = (known after apply)
+ publicly_accessible = true
+ region = "ap-northeast-1"
+ security_group_ids = (known after apply)
+ subnet_ids = (known after apply)
+ tags = {
+ "Project" = "iceberg-lab"
}
+ tags_all = {
+ "Project" = "iceberg-lab"
}
+ track_name = (known after apply)
+ workgroup_id = (known after apply)
+ workgroup_name = "iceberg-lab-wg"
+ config_parameter (known after apply)
+ price_performance_target (known after apply)
}
Plan: 5 to add, 0 to change, 0 to destroy.
───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
Note: You didn't use the -out option to save this plan, so Terraform can't guarantee to take exactly these actions if
you run "terraform apply" now.
terraform applyを実行します。
C:\s3-tables-iceberg-lab > terraform apply
module.s3_tables.data.aws_caller_identity.current: Reading...
module.s3_landing.data.aws_caller_identity.current: Reading...
module.step_functions.data.aws_caller_identity.current: Reading...
module.glue.data.aws_caller_identity.current: Reading...
module.iam.aws_iam_role.glue: Refreshing state... [id=iceberg-lab-glue-role]
module.iam.aws_iam_role.eventbridge: Refreshing state... [id=iceberg-lab-eventbridge-role]
module.iam.aws_iam_role.step_functions: Refreshing state... [id=iceberg-lab-sfn-role]
module.step_functions.aws_sns_topic.pipeline_notification: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccoundId}:iceberg-lab-pipeline-notification]
module.s3_tables.aws_s3tables_table_bucket.main: Refreshing state... [name=iceberg-lab]
module.s3_tables.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.s3_landing.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.s3_landing.aws_s3_bucket.landing: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.glue.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.glue.aws_s3_bucket.glue_scripts: Refreshing state... [id=iceberg-lab-glue-scripts-{AccoundId}]
module.step_functions.data.aws_caller_identity.current: Read complete after 0s [id={AccoundId}]
module.step_functions.aws_sns_topic_subscription.email: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccoundId}:iceberg-lab-pipeline-notification:e0e63b7a-82f3-44ea-8a59-ba38e5f2af88]
module.s3_tables.aws_s3tables_namespace.analytics: Refreshing state...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Reading...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Read complete after 0s [id=3931547792]
module.s3_tables.aws_s3tables_table_bucket_policy.main: Refreshing state...
module.s3_tables.aws_s3tables_table.orders: Refreshing state... [name=orders]
module.s3_tables.aws_s3tables_table.customers: Refreshing state... [name=customers]
module.s3_landing.aws_s3_bucket_notification.landing_eventbridge: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.s3_landing.aws_s3_bucket_versioning.landing: Refreshing state... [id=iceberg-lab-landing-{AccoundId}]
module.s3_landing.aws_cloudwatch_event_rule.s3_csv_arrived: Refreshing state... [id=iceberg-lab-csv-arrived]
module.glue.aws_s3_object.glue_script: Refreshing state... [id=iceberg-lab-glue-scripts-{AccoundId}/scripts/glue_csv_to_iceberg.py]
module.iam.aws_iam_role_policy_attachment.glue_service: Refreshing state... [id=iceberg-lab-glue-role/arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole]
module.iam.aws_iam_role_policy.glue_data_access: Refreshing state... [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
module.glue.aws_glue_job.csv_to_iceberg: Refreshing state... [id=iceberg-lab-csv-to-iceberg]
module.iam.aws_iam_role_policy.sfn_glue_sns: Refreshing state... [id=iceberg-lab-sfn-role:glue-and-sns]
module.step_functions.aws_sfn_state_machine.etl_pipeline: Refreshing state... [id=arn:aws:states:ap-northeast-1:{AccoundId}:stateMachine:iceberg-lab-etl-pipeline]
module.iam.aws_iam_role_policy.eventbridge_start_sfn: Refreshing state... [id=iceberg-lab-eventbridge-role:start-step-functions]
module.s3_landing.aws_cloudwatch_event_target.step_functions: Refreshing state... [id=iceberg-lab-csv-arrived-terraform-20260225153517622700000001]
Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the
following symbols:
+ create
Terraform will perform the following actions:
# module.iam.aws_iam_role.redshift will be created
+ resource "aws_iam_role" "redshift" {
+ arn = (known after apply)
+ assume_role_policy = jsonencode(
{
+ Statement = [
+ {
+ Action = "sts:AssumeRole"
+ Effect = "Allow"
+ Principal = {
+ Service = "redshift.amazonaws.com"
}
},
]
+ Version = "2012-10-17"
}
)
+ create_date = (known after apply)
+ force_detach_policies = false
+ id = (known after apply)
+ managed_policy_arns = (known after apply)
+ max_session_duration = 3600
+ name = "iceberg-lab-redshift-role"
+ name_prefix = (known after apply)
+ path = "/"
+ tags_all = (known after apply)
+ unique_id = (known after apply)
+ inline_policy (known after apply)
}
# module.iam.aws_iam_role_policy.redshift_lakehouse will be created
+ resource "aws_iam_role_policy" "redshift_lakehouse" {
+ id = (known after apply)
+ name = "lakehouse-access"
+ name_prefix = (known after apply)
+ policy = jsonencode(
{
+ Statement = [
+ {
+ Action = [
+ "glue:GetDatabase",
+ "glue:GetDatabases",
+ "glue:GetTable",
+ "glue:GetTables",
+ "glue:GetPartitions",
+ "glue:GetCatalog",
+ "glue:GetCatalogs",
+ "glue:CreateTable",
+ "glue:UpdateTable",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "s3tables:*",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:ListBucket",
+ "s3:GetBucketLocation",
]
+ Effect = "Allow"
+ Resource = "*"
},
+ {
+ Action = [
+ "lakeformation:GetDataAccess",
]
+ Effect = "Allow"
+ Resource = "*"
},
]
+ Version = "2012-10-17"
}
)
+ role = (known after apply)
}
# module.redshift.aws_redshiftserverless_namespace.main will be created
+ resource "aws_redshiftserverless_namespace" "main" {
+ admin_password_secret_arn = (known after apply)
+ admin_password_secret_kms_key_id = (known after apply)
+ admin_user_password = (sensitive value)
+ admin_user_password_wo = (write-only attribute)
+ admin_username = (sensitive value)
+ arn = (known after apply)
+ db_name = "dev"
+ iam_roles = (known after apply)
+ id = (known after apply)
+ kms_key_id = (known after apply)
+ namespace_id = (known after apply)
+ namespace_name = "iceberg-lab-ns"
+ region = "ap-northeast-1"
+ tags = {
+ "Project" = "iceberg-lab"
}
+ tags_all = {
+ "Project" = "iceberg-lab"
}
}
# module.redshift.aws_redshiftserverless_usage_limit.daily will be created
+ resource "aws_redshiftserverless_usage_limit" "daily" {
+ amount = 8
+ arn = (known after apply)
+ breach_action = "deactivate"
+ id = (known after apply)
+ period = "daily"
+ region = "ap-northeast-1"
+ resource_arn = (known after apply)
+ usage_type = "serverless-compute"
}
# module.redshift.aws_redshiftserverless_workgroup.main will be created
+ resource "aws_redshiftserverless_workgroup" "main" {
+ arn = (known after apply)
+ base_capacity = 4
+ endpoint = (known after apply)
+ id = (known after apply)
+ namespace_name = "iceberg-lab-ns"
+ port = (known after apply)
+ publicly_accessible = true
+ region = "ap-northeast-1"
+ security_group_ids = (known after apply)
+ subnet_ids = (known after apply)
+ tags = {
+ "Project" = "iceberg-lab"
}
+ tags_all = {
+ "Project" = "iceberg-lab"
}
+ track_name = (known after apply)
+ workgroup_id = (known after apply)
+ workgroup_name = "iceberg-lab-wg"
+ config_parameter (known after apply)
+ price_performance_target (known after apply)
}
Plan: 5 to add, 0 to change, 0 to destroy.
Do you want to perform these actions?
Terraform will perform the actions described above.
Only 'yes' will be accepted to approve.
Enter a value: yes
module.iam.aws_iam_role.redshift: Creating...
module.iam.aws_iam_role.redshift: Creation complete after 2s [id=iceberg-lab-redshift-role]
module.iam.aws_iam_role_policy.redshift_lakehouse: Creating...
module.redshift.aws_redshiftserverless_namespace.main: Creating...
module.redshift.aws_redshiftserverless_namespace.main: Creation complete after 0s [id=iceberg-lab-ns]
module.redshift.aws_redshiftserverless_workgroup.main: Creating...
module.iam.aws_iam_role_policy.redshift_lakehouse: Creation complete after 0s [id=iceberg-lab-redshift-role:lakehouse-access]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [00m10s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [00m20s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [00m30s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [00m40s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [00m50s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [01m00s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [01m10s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Still creating... [01m20s elapsed]
module.redshift.aws_redshiftserverless_workgroup.main: Creation complete after 1m26s [id=iceberg-lab-wg]
module.redshift.aws_redshiftserverless_usage_limit.daily: Creating...
module.redshift.aws_redshiftserverless_usage_limit.daily: Creation complete after 0s [id=57431a2c-1c31-45e0-9a73-2ac3e771c0bc]
Apply complete! Resources: 5 added, 0 changed, 0 destroyed.
リソースの作成が完了しました。
上記のようにSETで設定した環境変数はセッション限りの一時的なもの。システム環境変数に設定することで新たなセッションを作成時、同様の作業を避けることができる。
4RPU構成の制限事項
| 項目 | 制限 |
|---|---|
| ベースメモリ | 64GB（4 RPU × 16 GB） |
| 最大マネージドストレージ | 32TB |
| テーブルあたり最大カラム数 | 100カラム |
| VACUUM BOOST(テーブルの不要領域を解放・ソートするVACUUM操作を、専用の追加リソースを割り当てて高速化する) | 非対応（8RPU以上が必要） |
検証用途であれば4RPUで十分です。本番で100カラム超のテーブルを扱う場合は8RPU以上を検討してください。
Step 2：Lake FormationでRedshiftにアクセス権限を付与
RedshiftよりS3 Tablesにクエリを実行するためのカタログ、データベース、テーブルに対するアクセス権限をLake Formationで設定します。
まずはカタログへの権限を設定します。
| 項目 | 設定値 |
|---|---|
| Principals | Redshiftサービスロール(RedshiftからLake Formationにアクセス可能にするため。Redshiftがカタログを見れる状態にする。) |
| LF-Tags or catalog resources | Named Data Catalog resources |
| Catalogs | s3tablescatalog/<テーブルバケット名> |
| permissions | DESCRIBE |
次にデータベースへの権限を設定します。
| 項目 | 設定値 |
|---|---|
| Principals | 今回作成したRedshiftが使用するロール |
| LF-Tags or catalog resources | Named Data Catalog resources |
| Catalogs | s3tablescatalog/<テーブルバケット名> |
| Databases | analytics |
| permissions | DESCRIBE, CREATE_TABLE |
最後にテーブルへの権限を設定します。
| 項目 | 設定値 |
|---|---|
| Principals | 今回作成したRedshiftが使用するロール |
| LF-Tags or catalog resources | Named Data Catalog resources |
| Catalogs | s3tablescatalog/<テーブルバケット名> |
| Databases | analytics |
| permissions | SELECT, INSERT, DELETE, DESCRIBE, ALTER |
| Tables | orders |
3レベルでの権限付与が完了しました。
Step 3：RedshiftからIceberg テーブルをクエリ
3-1. Redshift Query Editor v2に接続
RedShiftのコンソール画面から、ワークグループを選択し、クエリエディタを開始します。
画面左のワークグループを選択すると、接続開始設定の画面が出るので、接続方法はfederated user、接続先のデータベースはdevとします。
外部データベースとして、Lake Formationに登録されたs3tablescatalogがマウントされていることを確認できます。また、s3tablescatalogに登録されたテーブルバケット、その中のデータベース、テーブルも確認できます。
3-2. Icebergテーブルへのクエリ
いくつかのクエリを実行してみます。
〇S3 TablesのIcebergテーブルを直接クエリ
SELECT
order_id,
customer_id,
order_date,
status
FROM
"iceberg-lab@s3tablescatalog"."analytics"."orders"
ORDER BY order_date DESC
LIMIT 10;
-- 実行結果
order_id customer_id order_date status
1008 3 2024-10-01 completed
1007 2 2024-09-01 completed
1005 4 2024-07-15 completed
1004 1 2024-07-01 completed
1003 3 2024-06-10 completed
1002 2 2024-06-05 completed
1001 1 2024-06-01 refunded
1014 2003 2024-01-17 pending
1011 2003 2024-01-17 pending
1014 2003 2024-01-17 pending
〇RedshiftローカルテーブルとIcebergテーブルの結合（ローカルにマスタ、Icebergにトランザクションデータ）
Redshiftのローカルデータベースdevにcustomersテーブルを作成、データを挿入します。
CREATE TABLE customers (
customer_id BIGINT,
customer_name VARCHAR(256),
email VARCHAR(256),
city VARCHAR(256),
country VARCHAR(256),
segment VARCHAR(256),
created_at TIMESTAMP
);
INSERT INTO customers VALUES (
1,
'Taro Yamada',
'taro.yamada@example.com',
'Tokyo',
'Japan',
'Consumer',
'2024-01-15 09:00:00'
);
クエリエディタでテーブルを確認できます。
RedshiftローカルテーブルcustomersとIcebergテーブルordersを結合し、データの参照を行います。
SELECT
o.order_id,
c.customer_name,
o.order_date,
o.status
FROM
"iceberg-lab@s3tablescatalog"."analytics"."orders" o
JOIN
dev.public.customers c ON o.customer_id = c.customer_id;
-- 実行結果
order_id customer_name order_date status
1001 Taro Yamada 2024-06-01 refunded
1004 Taro Yamada 2024-07-01 completed
AthenaとRedshiftで同じIcebergテーブルを参照できる。データのコピーは一切不要。Athenaはアドホック分析向き、Redshiftは複雑な結合や大量集計向き、と使い分けられる。
Step 4：RedshiftからIcebergテーブルへの書き込み
RedshiftはIcebergテーブルへの直接書き込みもサポートしています。
4-1. RedshiftからIcebergテーブルにINSERT
RedshiftのSQLで直接Icebergテーブルに書き込み、参照してみます。
INSERT INTO "iceberg-lab@s3tablescatalog"."analytics"."orders"
(order_id, customer_id, order_date, total_amount, status)
VALUES
(9001, 1, DATE '2026-02-28', 50000, 'completed'),
(9002, 2, DATE '2026-02-28', 32000, 'processing');
Icebergテーブルにデータが挿入されていることが確認できました。
SELECT
order_id,
customer_id,
order_date,
status
FROM
"iceberg-lab@s3tablescatalog"."analytics"."orders"
ORDER BY order_date DESC
LIMIT 10;
-- 実行結果
order_id customer_id order_date status
1008 3 2024-10-01 completed
1007 2 2024-09-01 completed
1005 4 2024-07-15 completed
1004 1 2024-07-01 completed
1003 3 2024-06-10 completed
1002 2 2024-06-05 completed
1001 1 2024-06-01 refunded
1014 2003 2024-01-17 pending
1011 2003 2024-01-17 pending
1014 2003 2024-01-17 pending
9001 1 2026-02-28 completed
9002 2 2026-02-28 processing
4-2. CTAS（集計結果を新しいIcebergテーブルとして保存）
Icebergテーブルから集計した日次集計結果をIcebergテーブルとして作成します。
CREATE TABLE "iceberg-lab@s3tablescatalog"."analytics"."daily_summary"
USING ICEBERG
AS
SELECT
order_date,
COUNT(*) AS order_count,
SUM(total_amount) AS total_amount,
CAST(AVG(total_amount) AS DECIMAL(12,2)) AS avg_amount,
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) AS completed_count
FROM
"iceberg-lab@s3tablescatalog"."analytics"."orders"
GROUP BY order_date;
クエリエディタでテーブルを確認できます。
S3のテーブルバケットのコンソール画面からも確認できます。
作成したテーブルは想定通りのデータを持っていました。
SELECT * FROM "iceberg-lab@s3tablescatalog"."analytics"."daily_summary"
ORDER BY order_date;
-- 実行結果
order_date order_count total_amount avg_amount completed_count
2024-01-15 3 294000.00 98000.00 3
2024-01-16 3 22500.00 7500.00 3
2024-01-17 3 34800.00 11600.00 0
2024-06-01 1 359.88 359.88 0
2024-06-05 1 599.97 599.97 1
2024-06-10 1 499.99 499.99 1
2024-07-01 1 719.76 719.76 1
2024-07-15 1 189.99 189.99 1
2024-09-01 1 849.96 849.96 1
2024-10-01 1 299.99 299.99 1
RedshiftのIceberg書き込みでサポートされるDMLはCREATE TABLE（CTAS含む）とINSERTのみであり、DELETE, UPDATE, MERGE, ALTER TABLEは未サポート。行レベルの更新・削除が必要な場合は、引き続きAthena / Glueなどを使用する。
Part B: Glue Data Quality品質ゲート
Step 5：Glue Data Qualityルールの設計
AWS Glue Data QualityはS3 Tables、Lake Formation管理のIcebergテーブルにも対応しています。ETLパイプラインに品質チェックを組み込みます。CSVデータが以下のルール全てを満たさない場合、Icebergテーブルへの投入をブロックします。
品質ルール設計
Rules = [
# 1. レコード数: 空ファイルでないこと
RowCount > 0,
# 2. order_id: 全行に値があること（NULL禁止）
IsComplete "order_id",
# 3. order_id: 重複がないこと
IsUnique "order_id",
# 4. total_amount: 全行に値があること
IsComplete "total_amount",
# 5. total_amount: 0以上であること（マイナス金額はNG）
ColumnValues "total_amount" >= 0,
# 6. status: 許可された値のみ
ColumnValues "status" in ["completed", "pending", "processing", "cancelled"],
# 7. order_date: 全行に値があること
IsComplete "order_date",
# 8. customer_id: 全行に値があること
IsComplete "customer_id"
]
ルール設計のポイント
| ルール種別 | 目的 | 失敗時の影響 |
|---|---|---|
| RowCount | 空ファイル検知 | Glue Jobの無駄な実行を防止 |
| IsComplete | NULLチェック | 下流のJOINやフィルタでの予期しない結果を防止 |
| IsUnique | 重複チェック | 集計の二重カウントを防止 |
| ColumnValues (範囲) | 値域チェック | 異常値の混入を防止 |
| ColumnValues (許可リスト) | カテゴリ値チェック | 未定義ステータスの混入を防止 |
Step 6：Glue Jobに品質チェックを組み込み
第2弾記事のGlue Jobスクリプト（scripts/glue_csv_to_iceberg.py）を拡張し、品質チェックを追加します。
scripts/glue_csv_to_iceberg.py（品質チェック追加版）
import sys
import json
import boto3
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from awsgluedq.transforms import EvaluateDataQuality # AWS Glue Data Quality
from pyspark.context import SparkContext
from pyspark.sql import SparkSession
# ── パラメータ取得 ──
args = getResolvedOptions(sys.argv, [
'JOB_NAME', 'SOURCE_PATH', 'TARGET_TABLE',
'CATALOG_ID', 'WAREHOUSE_PATH', 'SNS_TOPIC_ARN'
])
# ── Sparkセッション設定 ──
spark = (SparkSession.builder
.config("spark.sql.extensions",
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config("spark.sql.defaultCatalog", "s3tablescatalog")
.config("spark.sql.catalog.s3tablescatalog",
"org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.s3tablescatalog.catalog-impl",
"org.apache.iceberg.aws.glue.GlueCatalog")
.config("spark.sql.catalog.s3tablescatalog.glue.id", args['CATALOG_ID'])
.config("spark.sql.catalog.s3tablescatalog.warehouse", args['WAREHOUSE_PATH'])
.config("spark.sql.iceberg.handle-timestamp-without-timezone", "true")
.getOrCreate()
)
glueContext = GlueContext(spark.sparkContext)
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
sns_client = boto3.client('sns')
# ── Step 1: CSV読み込み ──
print(f"Reading CSV from: {args['SOURCE_PATH']}")
df = spark.read \
.option("header", "true") \
.option("inferSchema", "true") \
.csv(args['SOURCE_PATH'])
record_count = df.count()
print(f"Record count: {record_count}")
# DynamicFrame に変換（EvaluateDataQualityがDynamicFrameを受け取るため）
dyf = DynamicFrame.fromDF(df, glueContext, "incoming_orders")
# ── Step 2: DQDLでデータ品質ルールを定義 ──
DQDL_RULESET = """
Rules = [
RowCount > 0,
IsComplete "order_id",
IsUnique "order_id",
IsComplete "total_amount",
ColumnValues "total_amount" >= 0,
ColumnValues "status" in ["completed", "pending", "processing", "cancelled"],
IsComplete "order_date",
IsComplete "customer_id"
]
"""
# ── Step 3: EvaluateDataQualityで品質チェック実行 ──
dq_results = EvaluateDataQuality().process_rows(
frame=dyf,
ruleset=DQDL_RULESET,
publishing_options={
"dataQualityEvaluationContext": "orders_quality_check",
"enableDataQualityCloudWatchMetrics": True, # CloudWatchにメトリクス送信
"enableDataQualityResultsPublishing": True, # Glueコンソールに結果表示
},
additional_options={
"performanceTuning.caching": "CACHE_NOTHING"
}
)
# ルールごとの合否結果を取得
rule_outcomes_dyf = SelectFromCollection.apply(
dfc=dq_results,
key="ruleOutcomes",
transformation_ctx="ruleOutcomes"
)
rule_outcomes_df = rule_outcomes_dyf.toDF()
# ── Step 4: 品質スコア計算＆レポート出力 ──
total_rules = rule_outcomes_df.count()
passed_rules = rule_outcomes_df.filter(
rule_outcomes_df["Outcome"] == "Passed"
).count()
quality_score = (passed_rules / total_rules) * 100
print(f"\n{'='*50}")
print(f"DATA QUALITY REPORT")
print(f"{'='*50}")
print(f"Score: {quality_score:.1f}% ({passed_rules}/{total_rules} rules passed)")
rule_outcomes_df.select("Rule", "Outcome", "FailureReason").show(truncate=False)
print(f"{'='*50}\n")
# ── Step 5: 品質ゲート判定 ──
if quality_score < 100:
# 不合格：隔離フォルダに退避
quarantine_path = args['SOURCE_PATH'].replace('/raw/', '/quarantine/')
df.write.mode("overwrite").option("header", "true").csv(quarantine_path)
print(f"Quarantined data to: {quarantine_path}")
# 失敗ルール一覧を収集してSNS通知
failed_rules = rule_outcomes_df.filter(
rule_outcomes_df["Outcome"] == "Failed"
).select("Rule", "FailureReason").collect()
failure_detail = "\n".join(
f" - {row['Rule']}: {row['FailureReason']}" for row in failed_rules
)
sns_client.publish(
TopicArn=args['SNS_TOPIC_ARN'],
Subject="Data Quality Check FAILED",
Message=(
f"Data Quality FAILED for {args['SOURCE_PATH']}\n"
f"Score: {quality_score:.1f}%\n"
f"Failed rules:\n{failure_detail}"
)
)
# ジョブを失敗させる → Step FunctionsのCatchに遷移
raise Exception(f"Data Quality check failed: {quality_score:.1f}%")
# ── Step 6: 品質チェック合格 → Icebergテーブルに投入 ──
print("Data Quality PASSED. Writing to Iceberg table...")
df.createOrReplaceTempView("incoming_orders")
spark.sql(f"""
INSERT INTO {args['TARGET_TABLE']}
SELECT *
FROM incoming_orders
""")
print(f"Successfully wrote {record_count} records to {args['TARGET_TABLE']}")
# 合格通知
sns_client.publish(
TopicArn=args['SNS_TOPIC_ARN'],
Subject="Data Quality Check PASSED",
Message=(
f"File: {args['SOURCE_PATH']}\n"
f"Records: {record_count}\n"
f"Quality Score: {quality_score:.1f}%\n"
f"All {total_rules} rules passed."
)
)
job.commit()
Step 7：TerraformでGlue Data Qualityルールセットを定義
Glue Data Catalogに品質ルールセットを登録しておくと、コンソールからルールの確認・手動実行ができます。
modules/glue/main.tf
# 現在の実行アカウント情報を取得
data "aws_caller_identity" "current" {}
# gluescript格納用バケット
resource "aws_s3_bucket" "glue_scripts" {
bucket = "${var.project_name}-glue-scripts-${data.aws_caller_identity.current.account_id}"
}
# gluescriptをS3バケットへ配置
resource "aws_s3_object" "glue_script" {
bucket = aws_s3_bucket.glue_scripts.id
key = "scripts/glue_csv_to_iceberg.py"
source = "${path.module}/scripts/glue_csv_to_iceberg.py"
etag = filemd5("${path.module}/scripts/glue_csv_to_iceberg.py")
}
# gluejob
resource "aws_glue_job" "csv_to_iceberg" {
name = "${var.project_name}-csv-to-iceberg"
role_arn = var.glue_iamrole_arn
command {
script_location = "s3://${aws_s3_bucket.glue_scripts.id}/scripts/glue_csv_to_iceberg.py"
python_version = "3"
}
glue_version = "5.0"
number_of_workers = 2
worker_type = "G.1X"
timeout = 60
default_arguments = {
"--job-language" = "python"
"--enable-metrics" = "true"
"--enable-continuous-cloudwatch-log" = "true"
"--datalake-formats" = "iceberg"
"--CATALOG_ID" = "${data.aws_caller_identity.current.account_id}:s3tablescatalog/${var.s3tables_bucket_name}"
"--WAREHOUSE_PATH" = "s3://${var.s3tables_bucket_name}/warehouse/"
}
}
################################以下今回追加箇所#####################################
locals {
orders_ruleset = join(", ", [
"Rules = [RowCount > 0",
"IsComplete \"order_id\"",
"IsUnique \"order_id\"",
"IsComplete \"total_amount\"",
"ColumnValues \"total_amount\" >= 0",
"ColumnValues \"status\" in [\"completed\",\"pending\",\"processing\",\"cancelled\"]",
"IsComplete \"order_date\"",
"IsComplete \"customer_id\"]"
])
}
resource "aws_glue_data_quality_ruleset" "orders" {
name = "${var.project_name}-orders-quality-rules"
description = "ordersテーブルのデータ品質ルール"
# ルールの対象テーブル（Glue Data Catalog）
target_table {
database_name = "analytics"
table_name = "orders"
# S3 Tablesカタログの場合、catalog_idを指定
catalog_id = "${data.aws_caller_identity.current.account_id}:s3tablescatalog/${var.s3tables_bucket_name}"
}
ruleset = local.orders_ruleset # localで定義したローカル変数を使用
}
Step 8：Glue Jobの設定を更新
modules/glue/main.tf
SNS Topic ARNをGlue Jobに渡します。
resource "aws_glue_job" "csv_to_iceberg" {
name = "${var.project_name}-csv-to-iceberg"
role_arn = aws_iam_role.glue.arn
command {
script_location = "s3://${aws_s3_bucket.landing.id}/scripts/glue_csv_to_iceberg.py"
python_version = "3"
}
glue_version = "5.0"
number_of_workers = 2
worker_type = "G.1X"
timeout = 60
default_arguments = {
"--job-language" = "python"
"--enable-metrics" = "true"
"--enable-continuous-cloudwatch-log" = "true"
"--datalake-formats" = "iceberg"
"--CATALOG_ID" = "${data.aws_caller_identity.current.account_id}:s3tablescatalog/${var.s3tables_bucket_name}"
"--WAREHOUSE_PATH" = "s3://${var.s3tables_bucket_name}/warehouse/"
"--SNS_TOPIC_ARN" = var.pipeline_notification_sns_topic_arn
}
}
modules/glue/variables.tf
main.tfにて使用する変数を定義します。
variable "project_name" {
description = "プロジェクト名"
type = string
default = "iceberg-lab"
}
variable "glue_iamrole_arn" {
description = "glue用IAMロールARN"
type = string
}
variable "s3tables_bucket_name" {
description = "S3テーブル用バケット名"
type = string
}
################################以下今回追加箇所#####################################
variable "pipeline_notification_sns_topic_arn" {
description = "ETLパイプライン通知用SNSトピックARN"
type = string
}
modules/iam/main.tf
GlueロールにSNS発行権限、ランディング用バケットへのオブジェクト作成権限を追加します。
#glue用IAMポリシー
resource "aws_iam_role_policy" "glue_data_access" {
name = "${var.project_name}-glue-data-access"
role = aws_iam_role.glue.id
# 生データランディング用バケット、S3Tables用バケット、Lakeformation、Glueに対する権限
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Sid = "S3LandingAccess"
Effect = "Allow"
Action = ["s3:GetObject", "s3:ListBucket", "s3:PutObject"] # 変更箇所
Resource = [
var.landing_bucket_arn,
"${var.landing_bucket_arn}/*",
var.scripts_bucket_arn,
"${var.scripts_bucket_arn}/*",
]
},
{
Sid = "S3TablesAccess"
Effect = "Allow"
Action = ["s3tables:*"]
Resource = [
var.s3tables_bucket_arn,
"${var.s3tables_bucket_arn}/*"
]
},
{
Sid = "GlueAndLakeFormation"
Effect = "Allow"
Action = ["glue:*", "lakeformation:GetDataAccess"]
Resource = ["*"]
},
################################以下今回追加箇所#####################################
{
Sid = "GlueSnsPublish"
Effect = "Allow"
Action = "sns:Publish"
Resource = var.pipeline_notification_sns_topic_arn
}
]
})
}
main.tf
モジュール呼び出しを行うmain.tfに変更を加えます。
# Glue関連
module "glue" {
source = "./modules/glue/"
project_name = var.project_name
s3tables_bucket_name = module.s3_tables.s3tables_bucket_name
glue_iamrole_arn = module.iam.glue_iamrole_arn
# 変更箇所
pipeline_notification_sns_topic_arn = module.step_functions.pipeline_notification_sns_topic_arn
}
デプロイ
terraform planを実行します。
C:\s3-tables-iceberg-lab > terraform plan
module.step_functions.data.aws_caller_identity.current: Reading...
module.glue.data.aws_caller_identity.current: Reading...
module.s3_tables.data.aws_caller_identity.current: Reading...
module.s3_landing.data.aws_caller_identity.current: Reading...
module.iam.aws_iam_role.redshift: Refreshing state... [id=iceberg-lab-redshift-role]
module.iam.aws_iam_role.step_functions: Refreshing state... [id=iceberg-lab-sfn-role]
module.iam.aws_iam_role.eventbridge: Refreshing state... [id=iceberg-lab-eventbridge-role]
module.step_functions.aws_sns_topic.pipeline_notification: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification]
module.iam.aws_iam_role.glue: Refreshing state... [id=iceberg-lab-glue-role]
module.s3_tables.aws_s3tables_table_bucket.main: Refreshing state... [name=iceberg-lab]
module.s3_landing.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.s3_landing.aws_s3_bucket.landing: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.glue.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.glue.aws_s3_bucket.glue_scripts: Refreshing state... [id=iceberg-lab-glue-scripts-{AccountId}]
module.s3_tables.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.step_functions.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.step_functions.aws_sns_topic_subscription.email: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification:e0e63b7a-82f3-44ea-8a59-ba38e5f2af88]
module.s3_tables.aws_s3tables_namespace.analytics: Refreshing state...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Reading...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Read complete after 0s [id=3931547792]
module.s3_tables.aws_s3tables_table_bucket_policy.main: Refreshing state...
module.s3_tables.aws_s3tables_table.customers: Refreshing state... [name=customers]
module.s3_tables.aws_s3tables_table.orders: Refreshing state... [name=orders]
module.s3_landing.aws_s3_bucket_versioning.landing: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.s3_landing.aws_cloudwatch_event_rule.s3_csv_arrived: Refreshing state... [id=iceberg-lab-csv-arrived]
module.s3_landing.aws_s3_bucket_notification.landing_eventbridge: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.glue.aws_s3_object.glue_script: Refreshing state... [id=iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py]
module.iam.aws_iam_role_policy.redshift_lakehouse: Refreshing state... [id=iceberg-lab-redshift-role:lakehouse-access]
module.redshift.aws_redshiftserverless_namespace.main: Refreshing state... [id=iceberg-lab-ns]
module.iam.aws_iam_role_policy_attachment.glue_service: Refreshing state... [id=iceberg-lab-glue-role/arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole]
module.iam.aws_iam_role_policy.glue_data_access: Refreshing state... [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
module.glue.aws_glue_job.csv_to_iceberg: Refreshing state... [id=iceberg-lab-csv-to-iceberg]
module.redshift.aws_redshiftserverless_workgroup.main: Refreshing state... [id=iceberg-lab-wg]
module.iam.aws_iam_role_policy.sfn_glue_sns: Refreshing state... [id=iceberg-lab-sfn-role:glue-and-sns]
module.step_functions.aws_sfn_state_machine.etl_pipeline: Refreshing state... [id=arn:aws:states:ap-northeast-1:{AccountId}:stateMachine:iceberg-lab-etl-pipeline]
module.redshift.aws_redshiftserverless_usage_limit.daily: Refreshing state... [id=57431a2c-1c31-45e0-9a73-2ac3e771c0bc]
module.iam.aws_iam_role_policy.eventbridge_start_sfn: Refreshing state... [id=iceberg-lab-eventbridge-role:start-step-functions]
module.s3_landing.aws_cloudwatch_event_target.step_functions: Refreshing state... [id=iceberg-lab-csv-arrived-terraform-20260225153517622700000001]
Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the
following symbols:
+ create
~ update in-place
Terraform will perform the following actions:
# module.glue.aws_glue_data_quality_ruleset.orders will be created
+ resource "aws_glue_data_quality_ruleset" "orders" {
+ arn = (known after apply)
+ created_on = (known after apply)
+ description = "ordersテーブルのデータ品質ルール"
+ id = (known after apply)
+ last_modified_on = (known after apply)
+ name = "iceberg-lab-orders-quality-rules"
+ recommendation_run_id = (known after apply)
+ region = "ap-northeast-1"
+ ruleset = "Rules = [RowCount > 0, IsComplete \"order_id\", IsUnique \"order_id\", IsComplete \"total_amount\", ColumnValues \"total_amount\" >= 0, ColumnValues \"status\" in [\"completed\",\"pending\",\"processing\",\"cancelled\"], IsComplete \"order_date\", IsComplete \"customer_id\"]"
+ tags_all = (known after apply)
+ target_table {
+ catalog_id = "{AccountId}:s3tablescatalog/iceberg-lab"
+ database_name = "analytics"
+ table_name = "orders"
}
}
# module.glue.aws_glue_job.csv_to_iceberg will be updated in-place
~ resource "aws_glue_job" "csv_to_iceberg" {
~ default_arguments = {
+ "--SNS_TOPIC_ARN" = "arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification"
# (6 unchanged elements hidden)
}
id = "iceberg-lab-csv-to-iceberg"
name = "iceberg-lab-csv-to-iceberg"
# (17 unchanged attributes hidden)
# (2 unchanged blocks hidden)
}
# module.glue.aws_s3_object.glue_script will be updated in-place
~ resource "aws_s3_object" "glue_script" {
~ etag = "df9c58f8917007d62a5b4d2fd36d4b23" -> "f5ad76623540211725c61d05595732db"
id = "iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py"
tags = {}
+ version_id = (known after apply)
# (25 unchanged attributes hidden)
}
# module.iam.aws_iam_role_policy.glue_data_access will be updated in-place
~ resource "aws_iam_role_policy" "glue_data_access" {
id = "iceberg-lab-glue-role:iceberg-lab-glue-data-access"
name = "iceberg-lab-glue-data-access"
~ policy = jsonencode(
~ {
~ Statement = [
# (2 unchanged elements hidden)
{
Action = [
"glue:*",
"lakeformation:GetDataAccess",
]
Effect = "Allow"
Resource = [
"*",
]
Sid = "GlueAndLakeFormation"
},
+ {
+ Action = "sns:Publish"
+ Effect = "Allow"
+ Resource = "arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification"
+ Sid = "GlueSnsPublish"
},
]
# (1 unchanged attribute hidden)
}
)
# (2 unchanged attributes hidden)
}
Plan: 1 to add, 3 to change, 0 to destroy.
───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
Note: You didn't use the -out option to save this plan, so Terraform can't guarantee to take exactly these actions if
you run "terraform apply" now.
terraform applyを実行します。
C:\s3-tables-iceberg-lab > terraform apply
module.step_functions.data.aws_caller_identity.current: Reading...
module.glue.data.aws_caller_identity.current: Reading...
module.s3_landing.data.aws_caller_identity.current: Reading...
module.s3_tables.data.aws_caller_identity.current: Reading...
module.iam.aws_iam_role.redshift: Refreshing state... [id=iceberg-lab-redshift-role]
module.iam.aws_iam_role.step_functions: Refreshing state... [id=iceberg-lab-sfn-role]
module.iam.aws_iam_role.glue: Refreshing state... [id=iceberg-lab-glue-role]
module.step_functions.aws_sns_topic.pipeline_notification: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification]
module.iam.aws_iam_role.eventbridge: Refreshing state... [id=iceberg-lab-eventbridge-role]
module.s3_tables.aws_s3tables_table_bucket.main: Refreshing state... [name=iceberg-lab]
module.step_functions.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.s3_tables.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.glue.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.glue.aws_s3_bucket.glue_scripts: Refreshing state... [id=iceberg-lab-glue-scripts-{AccountId}]
module.s3_landing.data.aws_caller_identity.current: Read complete after 0s [id={AccountId}]
module.s3_landing.aws_s3_bucket.landing: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.step_functions.aws_sns_topic_subscription.email: Refreshing state... [id=arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification:e0e63b7a-82f3-44ea-8a59-ba38e5f2af88]
module.s3_tables.aws_s3tables_namespace.analytics: Refreshing state...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Reading...
module.s3_tables.data.aws_iam_policy_document.table_bucket: Read complete after 0s [id=3931547792]
module.s3_tables.aws_s3tables_table_bucket_policy.main: Refreshing state...
module.s3_tables.aws_s3tables_table.orders: Refreshing state... [name=orders]
module.s3_tables.aws_s3tables_table.customers: Refreshing state... [name=customers]
module.s3_landing.aws_s3_bucket_versioning.landing: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.s3_landing.aws_s3_bucket_notification.landing_eventbridge: Refreshing state... [id=iceberg-lab-landing-{AccountId}]
module.s3_landing.aws_cloudwatch_event_rule.s3_csv_arrived: Refreshing state... [id=iceberg-lab-csv-arrived]
module.glue.aws_s3_object.glue_script: Refreshing state... [id=iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py]
module.iam.aws_iam_role_policy_attachment.glue_service: Refreshing state... [id=iceberg-lab-glue-role/arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole]
module.iam.aws_iam_role_policy.glue_data_access: Refreshing state... [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
module.glue.aws_glue_job.csv_to_iceberg: Refreshing state... [id=iceberg-lab-csv-to-iceberg]
module.iam.aws_iam_role_policy.redshift_lakehouse: Refreshing state... [id=iceberg-lab-redshift-role:lakehouse-access]
module.redshift.aws_redshiftserverless_namespace.main: Refreshing state... [id=iceberg-lab-ns]
module.redshift.aws_redshiftserverless_workgroup.main: Refreshing state... [id=iceberg-lab-wg]
module.iam.aws_iam_role_policy.sfn_glue_sns: Refreshing state... [id=iceberg-lab-sfn-role:glue-and-sns]
module.step_functions.aws_sfn_state_machine.etl_pipeline: Refreshing state... [id=arn:aws:states:ap-northeast-1:{AccountId}:stateMachine:iceberg-lab-etl-pipeline]
module.redshift.aws_redshiftserverless_usage_limit.daily: Refreshing state... [id=57431a2c-1c31-45e0-9a73-2ac3e771c0bc]
module.iam.aws_iam_role_policy.eventbridge_start_sfn: Refreshing state... [id=iceberg-lab-eventbridge-role:start-step-functions]
module.s3_landing.aws_cloudwatch_event_target.step_functions: Refreshing state... [id=iceberg-lab-csv-arrived-terraform-20260225153517622700000001]
Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the
following symbols:
+ create
~ update in-place
Terraform will perform the following actions:
# module.glue.aws_glue_data_quality_ruleset.orders will be created
+ resource "aws_glue_data_quality_ruleset" "orders" {
+ arn = (known after apply)
+ created_on = (known after apply)
+ description = "ordersテーブルのデータ品質ルール"
+ id = (known after apply)
+ last_modified_on = (known after apply)
+ name = "iceberg-lab-orders-quality-rules"
+ recommendation_run_id = (known after apply)
+ region = "ap-northeast-1"
+ ruleset = "Rules = [RowCount > 0, IsComplete \"order_id\", IsUnique \"order_id\", IsComplete \"total_amount\", ColumnValues \"total_amount\" >= 0, ColumnValues \"status\" in [\"completed\",\"pending\",\"processing\",\"cancelled\"], IsComplete \"order_date\", IsComplete \"customer_id\"]"
+ tags_all = (known after apply)
+ target_table {
+ catalog_id = "{AccountId}:s3tablescatalog/iceberg-lab"
+ database_name = "analytics"
+ table_name = "orders"
}
}
# module.glue.aws_glue_job.csv_to_iceberg will be updated in-place
~ resource "aws_glue_job" "csv_to_iceberg" {
~ default_arguments = {
+ "--SNS_TOPIC_ARN" = "arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification"
# (6 unchanged elements hidden)
}
id = "iceberg-lab-csv-to-iceberg"
name = "iceberg-lab-csv-to-iceberg"
# (17 unchanged attributes hidden)
# (2 unchanged blocks hidden)
}
# module.glue.aws_s3_object.glue_script will be updated in-place
~ resource "aws_s3_object" "glue_script" {
~ etag = "df9c58f8917007d62a5b4d2fd36d4b23" -> "f5ad76623540211725c61d05595732db"
id = "iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py"
tags = {}
+ version_id = (known after apply)
# (25 unchanged attributes hidden)
}
# module.iam.aws_iam_role_policy.glue_data_access will be updated in-place
~ resource "aws_iam_role_policy" "glue_data_access" {
id = "iceberg-lab-glue-role:iceberg-lab-glue-data-access"
name = "iceberg-lab-glue-data-access"
~ policy = jsonencode(
~ {
~ Statement = [
# (2 unchanged elements hidden)
{
Action = [
"glue:*",
"lakeformation:GetDataAccess",
]
Effect = "Allow"
Resource = [
"*",
]
Sid = "GlueAndLakeFormation"
},
+ {
+ Action = "sns:Publish"
+ Effect = "Allow"
+ Resource = "arn:aws:sns:ap-northeast-1:{AccountId}:iceberg-lab-pipeline-notification"
+ Sid = "GlueSnsPublish"
},
]
# (1 unchanged attribute hidden)
}
)
# (2 unchanged attributes hidden)
}
Plan: 1 to add, 3 to change, 0 to destroy.
Do you want to perform these actions?
Terraform will perform the actions described above.
Only 'yes' will be accepted to approve.
Enter a value: yes
module.glue.aws_glue_data_quality_ruleset.orders: Creating...
module.iam.aws_iam_role_policy.glue_data_access: Modifying... [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
module.glue.aws_s3_object.glue_script: Modifying... [id=iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py]
module.glue.aws_glue_job.csv_to_iceberg: Modifying... [id=iceberg-lab-csv-to-iceberg]
module.glue.aws_s3_object.glue_script: Modifications complete after 1s [id=iceberg-lab-glue-scripts-{AccountId}/scripts/glue_csv_to_iceberg.py]
module.glue.aws_glue_job.csv_to_iceberg: Modifications complete after 1s [id=iceberg-lab-csv-to-iceberg]
module.glue.aws_glue_data_quality_ruleset.orders: Creation complete after 1s [id=iceberg-lab-orders-quality-rules]
module.iam.aws_iam_role_policy.glue_data_access: Modifications complete after 1s [id=iceberg-lab-glue-role:iceberg-lab-glue-data-access]
Apply complete! Resources: 1 added, 3 changed, 0 destroyed.
リソースの作成が完了しました。
Step 9：動作確認
Data Qualityを用いて設定したデータの品質チェックの動作を確認します。
9-1. 正常データのテスト
まずは正常なデータを用いた動作確認を行います。
品質チェック合格データ(good_orders.csv)
order_id,customer_id,order_date,product_name,quantity,unit_price,total_amount,status,region,created_at,discount_rate,note
2001,1,2024-06-01,Cloud Storage 1TB,12,29.99,359.88,completed,ap-northeast-1,2024-06-01 10:00:00.000000,0.00,
2002,2,2024-06-05,Compute Instance m5.xl,3,199.99,599.97,completed,ap-northeast-1,2024-06-05 11:30:00.000000,0.00,
2003,3,2024-06-10,Database RDS,1,499.99,499.99,completed,us-east-1,2024-06-10 09:15:00.000000,0.00,
上記CSVファイルをS3にアップロードします。
パイプラインが起動され、ETLジョブが正常に実行されました。
品質チェックの結果とETLパイプライン成功のメールがSNSで送信されています。
また、glueの出力ログ、コンソール画面から品質チェックの結果を詳細に確認できます。
Reading CSV from: s3://iceberg-lab-landing-{AccountId}/raw/orders/good-orders.csv
Record count: 3
/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py:147: UserWarning: DataFrame constructor is internal. Do not directly use it.
==================================================
DATA QUALITY REPORT
==================================================
Score: 100.0% (8/8 rules passed)
+-------------------------------------------------------------------------+-------+-------------+
|Rule |Outcome|FailureReason|
+-------------------------------------------------------------------------+-------+-------------+
|RowCount > 0 |Passed |NULL |
|IsComplete "order_id" |Passed |NULL |
|IsUnique "order_id" |Passed |NULL |
|IsComplete "total_amount" |Passed |NULL |
|ColumnValues "total_amount" >= 0 |Passed |NULL |
|ColumnValues "status" in ["completed","pending","processing","cancelled"]|Passed |NULL |
|IsComplete "order_date" |Passed |NULL |
|IsComplete "customer_id" |Passed |NULL |
+-------------------------------------------------------------------------+-------+-------------+
==================================================
Data Quality PASSED. Writing to Iceberg table...
Successfully wrote 3 records to analytics.orders
Running autoDebugger shutdown hook.
パイプライン実行前と実行後のデータを比較し、品質チェックに合格したデータがIcebergテーブルに挿入されていることを確認します。
実行前のテーブルデータ
select * from orders;
-- 実行結果
order_id customer_id order_date product_name quantity unit_price total_amount status region created_at discount_rate note
1001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 refunded ap-northeast-1 2024-06-01 10:00:00.000000 返金処理済
1002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000
1003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000
実行後のデータ
select * from orders;
-- 実行結果
order_id customer_id order_date product_name quantity unit_price total_amount status region created_at discount_rate note
1001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 refunded ap-northeast-1 2024-06-01 10:00:00.000000 返金処理済
1002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000
1003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000
2001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 completed ap-northeast-1 2024-06-01 10:00:00.000000 0.00
2002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000 0.00
2003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000 0.00
CSVのデータが挿入されていました。
9-2. 不正データのテスト
次には不正なデータを用いた動作確認を行います。品質チェックで100%でない場合、エラーとなり、ETL処理を行わずS3バケットの隔離フォルダに格納されます。
品質チェック不合格データ（重複order_id + マイナス金額 + 不正ステータス）(bad_orders.csv)
order_id,customer_id,order_date,product_name,quantity,unit_price,total_amount,status,region,created_at,discount_rate,note
4001,1,2026-02-28,Cloud Storage 1TB,1,15000.00,15000.00,completed,ap-northeast-1,2026-02-28 10:00:00.000000,0.00,
4001,2,2026-02-28,Compute Instance,1,500.00,-500.00,unknown_status,ap-northeast-1,2026-02-28 11:00:00.000000,0.00,
4003,,2026-02-28,Database RDS,1,8000.00,8000.00,pending,us-east-1,2026-02-28 09:00:00.000000,0.00,
上記CSVファイルをS3にアップロードします。
パイプラインが起動され、ETLジョブが失敗しました。
品質チェックの結果とETLパイプライン失敗のメールがSNSで送信されています。
glueの出力ログ、コンソール画面から品質チェックの結果を詳細に確認します。50%の割合でルールを合格しています。
Reading CSV from: s3://iceberg-lab-landing-{AccountId}/raw/orders/bad-orders.csv
Record count: 3
/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py:147: UserWarning: DataFrame constructor is internal. Do not directly use it.
==================================================
DATA QUALITY REPORT
==================================================
Score: 50.0% (4/8 rules passed)
+-------------------------------------------------------------------------+-------+-------------------------------------------------------------------+
|Rule |Outcome|FailureReason |
+-------------------------------------------------------------------------+-------+-------------------------------------------------------------------+
|RowCount > 0 |Passed |NULL |
|IsComplete "order_id" |Passed |NULL |
|IsUnique "order_id" |Failed |Value: 0.3333333333333333 does not meet the constraint requirement!|
|IsComplete "total_amount" |Passed |NULL |
|ColumnValues "total_amount" >= 0 |Failed |Value: -500.0 does not meet the constraint requirement! |
|ColumnValues "status" in ["completed","pending","processing","cancelled"]|Failed |Value: 0.6666666666666666 does not meet the constraint requirement!|
|IsComplete "order_date" |Passed |NULL |
|IsComplete "customer_id" |Failed |Value: 0.6666666666666666 does not meet the constraint requirement!|
+-------------------------------------------------------------------------+-------+-------------------------------------------------------------------+
==================================================
Running autoDebugger shutdown hook.
S3の隔離フォルダに不正データファイルが配置されています。
パイプライン実行前と実行後のデータを比較し、品質チェックに不合格となったデータがIcebergテーブルに挿入されていないことを確認します。
実行前のテーブルデータ
select * from orders;
-- 実行結果
order_id customer_id order_date product_name quantity unit_price total_amount status region created_at discount_rate note
1001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 refunded ap-northeast-1 2024-06-01 10:00:00.000000 返金処理済
1002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000
1003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000
2001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 completed ap-northeast-1 2024-06-01 10:00:00.000000 0.00
2002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000 0.00
2003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000 0.00
実行後のデータ
select * from orders;
-- 実行結果
order_id customer_id order_date product_name quantity unit_price total_amount status region created_at discount_rate note
1001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 refunded ap-northeast-1 2024-06-01 10:00:00.000000 返金処理済
1002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000
1003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000
2001 1 2024-06-01 Cloud Storage 1TB 12 29.99 359.88 completed ap-northeast-1 2024-06-01 10:00:00.000000 0.00
2002 2 2024-06-05 Compute Instance m5.xl 3 199.99 599.97 completed ap-northeast-1 2024-06-05 11:30:00.000000 0.00
2003 3 2024-06-10 Database RDS 1 499.99 499.99 completed us-east-1 2024-06-10 09:15:00.000000 0.00
差異がないことを確認できました。
まとめ
今回Redshift Serverlessでのレイクハウスクエリ、Glue Data Qualityでの品質検査を検証することができました。以下のように第1弾、第2弾の記事で実装した構成を更に拡張し、レイクハウスとして実際のユースケースに近いものを構築しました。
| 観点 | 第1弾 | 第2弾 | 第3弾（本記事） |
|---|---|---|---|
| データストレージ | S3 Tables + Iceberg | ─ | ─ |
| クエリエンジン | Athena | ─ | Redshift Serverless |
| ETL | 手動Glue Job | EventBridge → Step Functions → Glue | Data Quality品質チェック |
| 通知 | ─ | SNS（成功/失敗） | SNS（品質レポート付き） |
| IaC | Terraform | Terraform | Terraform |
この記事が皆様のお役に立てば幸いです。質問やフィードバックがあれば、コメント欄でお気軽にお寄せください！