背景

ゴリゴリ系エンジニアの生みの親、pageoです。
AWSのサーバレスETLパイプライン管理サービスであるGlue Workflowsに関する日本語の情報が少なかったので、記事にしてまとめてみました。
ただサービスの特徴を説明しても面白くないので、Terraformを用いて各Glueサービス(Jobs/Crawlersなど)を一通り実装し、Terraform Registryで公開しています

Github Repositoryは以下です(スター頂けるとpageo喜びます)

「Terraform　Registryとはなんぞや?」「Terraform　Registryのmoduleでどうやって公開するんや？」という方は、以下の記事で詳しく説明しているのでそちらを参照ください。(ハート頂けるとpageo喜びます)

Glueとは

Glueをざっくり説明

Glueをざっくり一言で表すと「サーバーレスなデータ統合サービス」です。
"サーバレスなETLサービス"と表現しても良いのですが、Glue Data CatalogなどETL以外にもサービスとしての守備範囲が広いので、"サーバーレスなデータ統合サービス"とさせていただきました。

Glue Workflowsとは

各Glueサービス(Jobs/Crawlers/Triggersなど)を使用した一連の複雑なETLジョブを管理/可視化できるサービスです。

以下はGlue Workflowの可視化イメージです

参照: 公式サイト

Glue 各種サービスについて

Glue Triggers
- Glue JobsやGlue Crawlersを開始するサービスです
- トリガーの種類にはSCHEDULED, ON_DEMAND, CONDITIONAL, EVENTがあります
Glue Jobs
- ETLジョブを実行するサービスです
- ジョブコマンドの種類にはglueetl, pythonshell, gluestreamingがあります
Glue Crawlers
- Glue Catalogにメタデータを書き込んでくれるサービスです
- データソースとしてS3やDynamoDBなどさまざまなデータストアをクロールできます
Glue Data Catalog
- Databases
  - Tablesを管理するサービスです
- Tables
  - Glue Jobsのデータソース or ターゲットとして使用可能な、データストアのメタデータを管理するサービスです
  - 手動での管理や、Glue Crawlersを使った自動管理も可能です
その他(この記事での実装対象外)
- Glue Blueprints
  - Glue Workflowsを簡単に作成/共有できるサービスです
  - (使い道がわからん、、、)
- Glue Studio
  - ETL ジョブを簡単に作成/実行/監視できるビジュアルインターフェースです
  - (Terraformで本格的にIaC化する前の試験実装としてよく使用しています)
- Glue Databrew
  - Glue Studioを非エンジニア向けにしたノンコードなビジュアルデータ処理サービス
  - Databrew自体がかなりゴツめのサービス(Glue本体の管理コンソールページと切り離されている)なので、使用するまでに必要な学習量が多い気がします
  - (いつか使いたい)

ここまでざっくりとAWS Glueについて説明してきましたが、さらに詳細な説明が欲しい方は公式サイトやClassmethodさんの記事に丁寧に整理されているので、そちらを参考いただければと思います。

前提

構築内容

それでは本題です
この記事で作成するAWS Resourceは以下です

Glue Workflows
Glue Triggers
Glue Jobs
Glue Clawlers
Glue Data Catalog

イメージ

構築イメージは以下です

想定するETL Workflowは以下です

EventBridgeによりスケジューリングされた時間でGlue Workflowを起動
EventBridgeと連携されたGlue TriggersからGlue Crawlersを起動
Glue CrawlersでS3 Bucket(Data Source)からGlue Data Catalogを更新
Glue Data Catalogの更新完了をトリガーに、Glue Jobsを起動
Glue JobsでS3 Bucket(Data Source)からS3 Bucket(Data Target)に結果を出力

構成

ディレクトリ構成は以下になります。

+- terraform-aws-glue-workflows/
    |
    +---- examples/
    |       +---- sample/
    |              +---- main.tf
    |              +---- provider.tf
    |              +---- locals.tf
    |              +---- README.md
    +---- .gitignore
    +---- catalog.tf
    +---- crawler.tf
    +---- job.tf
    +---- workflow.tf
    +---- variables.tf
    +---- outputs.tf
    +---- README.md
    +---- LICENSE

使い方

Github/Terraform RegistryにExamplesを載せているので詳細はそちらを参照ください

以下はmoduleの使用例です

sample.tf

module "glue_workflows" {
  source = "page-o/glue-workflows/aws"

  workflow = {
    name        = "${local.env}-${local.project}-glue-workflow"
    description = "Sample"
  }
  event_trigger = {
    name        = "${local.env}-${local.project}-event-trigger"
    description = "Sample"
  }
  crawler_trigger = {
    name        = "${local.env}-${local.project}-crawler-trigger"
    description = "Sample"
  }
  catalog = {
    database_name        = "${local.env}-${local.project}-database"
    database_description = "Sample"
    tables = [
      {
        name             = "hoges"
        description      = "Sample"
        storage_location = "s3://${module.source_bucket.name}/hoges/"
        ser_de_name      = "hoges"
      },
      {
        name             = "fugas"
        description      = "Sample"
        storage_location = "s3://${module.source_bucket.name}/fugas/"
        ser_de_name      = "fugas"
      }
    ]
  }
  crawler = {
    name                  = "${local.env}-${local.project}-crawler"
    description           = "Sample"
    role_arn              = module.workflow_role.arn
    catalog_target_tables = ["hoges", "fugas"]
  }
  jobs = [
    {
      name              = "${local.env}-${local.project}-hoges-job"
      description       = "Sample"
      script_object_key = local.job_script_object_key.hoge
      default_arguments = local.default_arguments
    },
    {
      name              = "${local.env}-${local.project}-fugas-job"
      description       = "Sample"
      script_object_key = local.job_script_object_key.fuga
      default_arguments = local.default_arguments
    }
  ]
  job_role_arn        = module.workflow_role.arn
  job_tmp_bucket_name = module.scripts_bucket.name
}

実装

以下に書くResourceの実装について説明していきます。

基本的に、forを用いてコードの使い回しわ避けるなどDRY原則に基づいて実装しています
TerraformにおけるDRY原則の実践については、以下の記事で詳しく説明しているので参照ください。(ハート頂けるとpageo喜びます)

まずGlue Data Catalogですが、単純にDatabaseとTableのResourceを定義しているだけです
データソースがS3であることを前提として実装していますが、データソースによってはstorage_descriptorパラメータの書き方が若干変わると思われます

catalog.tf

resource "aws_glue_catalog_database" "main" {
  name        = var.catalog.database_name
  description = var.catalog.database_description
}

resource "aws_glue_catalog_table" "main" {
  count = length(var.catalog.tables)

  database_name = aws_glue_catalog_database.main.name
  name          = element(var.catalog.tables.*.name, count.index)
  description   = element(var.catalog.tables.*.description, count.index)
  table_type    = element(var.catalog.tables.*.table_type, count.index)

  storage_descriptor {
    location      = element(var.catalog.tables.*.storage_location, count.index)
    input_format  = element(var.catalog.tables.*.input_format, count.index)
    output_format = element(var.catalog.tables.*.output_format, count.index)

    ser_de_info {
      name                  = element(var.catalog.tables.*.ser_de_name, count.index)
      serialization_library = element(var.catalog.tables.*.serialization_library, count.index)

      parameters = {
        "serialization.format" = 1
      }
    }
  }

  lifecycle {
    ignore_changes = [parameters, storage_descriptor, owner]
  }
}

Glue Crawlerも同じく単純な定義です
ここも、データソースがGlue Catalogであることを前提として実装しているので、他のデータソースを使用する場合はcatalog_targetパラメータをs3_target/dynamodb_target/jdbc_target/mongodb_targetなどに変更する必要があります
その辺りはTerraform公式ドキュメントを参考に実装してください

crawler.tf

resource "aws_glue_crawler" "main" {
  database_name = aws_glue_catalog_database.main.name
  name          = var.crawler.name
  description   = var.crawler.description
  role          = var.crawler.role_arn

  catalog_target {
    database_name = aws_glue_catalog_database.main.name
    tables        = var.crawler.catalog_target_tables
  }

  schema_change_policy {
    delete_behavior = "LOG"
    update_behavior = "UPDATE_IN_DATABASE"
  }

  lifecycle {
    ignore_changes = [configuration]
  }
}

続いてGlue Jobsです
ここも、Glue JobsのTypeがglueetlであることを前提に実装しています
他のType(pythonshell/gluestreaming)の場合にはmax_capacity/timeout/commandパラメータを修正する必要があります
Terraform公式ドキュメントを参考に実装しましょう

job.tf

resource "aws_glue_job" "main" {
  count = length(var.jobs)

  name              = element(var.jobs.*.name, count.index)
  description       = element(var.jobs.*.description, count.index)
  glue_version      = element(var.jobs.*.glue_version, count.index)
  execution_class   = element(var.jobs.*.execution_class, count.index)
  max_retries       = element(var.jobs.*.max_retries, count.index)
  timeout           = element(var.jobs.*.timeout, count.index)
  worker_type       = element(var.jobs.*.worker_type, count.index)
  number_of_workers = element(var.jobs.*.number_of_workers, count.index)

  role_arn = var.job_role_arn

  execution_property {
    max_concurrent_runs = element(var.jobs.*.max_concurrent_runs, count.index)
  }

  command {
    script_location = "s3://${var.job_tmp_bucket_name}/${element(var.jobs.*.script_object_key, count.index)}"
  }

  default_arguments = element(var.jobs.*.default_arguments, count.index)
}

最後にGlue Workflowsです
上記で実装してきたGlue Jobs/Glue Crawlers/Glue CatalogをここでWorkflowとして連結しています
aws_glue_trigger.eventの設定ですが、構築イメージで示した通りEventBridgeによるWorkflowのトリガーを想定しているので、typeをEVENTにしています
「EventBridgeによるWorkflowのトリガー」の実装例は、Terraform公式ドキュメントにも記載がないので要注意です
以下の引用/キャプチャのように、Terraform公式ドキュメントのtypeの説明にEVENTの記載がないのですが、EventBridgeによるGlue Workflowのトリガーの際はtypeにEVENTを設定する必要があります

(Required) The type of trigger. Valid values are CONDITIONAL, ON_DEMAND, and SCHEDULED.

workflow.tf

resource "aws_glue_workflow" "main" {
  name        = var.workflow.name
  description = var.workflow.description
}

resource "aws_glue_trigger" "event" {
  workflow_name = aws_glue_workflow.main.name
  name          = var.event_trigger.name
  description   = var.event_trigger.description
  type          = "EVENT"

  event_batching_condition {
    batch_size   = 1
    batch_window = 900
  }

  actions {
    crawler_name = aws_glue_crawler.main.name
  }
}

resource "aws_glue_trigger" "crawler" {
  workflow_name = aws_glue_workflow.main.name
  name          = var.crawler_trigger.name
  description   = var.crawler_trigger.description
  type          = "CONDITIONAL"

  predicate {
    logical = "ANY"
    conditions {
      crawler_name     = aws_glue_crawler.main.name
      crawl_state      = "SUCCEEDED"
      logical_operator = "EQUALS"
    }
  }

  dynamic "actions" {
    for_each = aws_glue_job.main

    content {
      job_name = actions.value.id
    }
  }
}

一応variables.tfも載せときます
Github Repositoryに全てのコードが載っているので、詳しくはこちらを参照ください

variables.tf

variable "workflow" {
  description = "Workflow"
  type = object({
    name        = string
    description = string
  })
}

variable "event_trigger" {
  description = "Event Trigger"
  type = object({
    name        = string
    description = string
  })
}

variable "crawler_trigger" {
  description = "Crawler Trigger"
  type = object({
    name        = string
    description = string
  })
}

variable "catalog" {
  description = "Catalog"
  type = object({
    database_name        = string
    database_description = string
    tables = list(object({
      name                  = string
      description           = string
      table_type            = optional(string, "EXTERNAL_TABLE")
      storage_location      = string
      input_format          = optional(string, "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat")
      output_format         = optional(string, "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat")
      ser_de_name           = string
      serialization_library = optional(string, "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
    }))
  })
}

variable "crawler" {
  description = "Crawler"
  type = object({
    name                  = string
    description           = string
    role_arn              = string
    catalog_target_tables = list(string)
  })
}

variable "jobs" {
  description = "Jobs"
  type = list(object({
    name                = string
    description         = string
    glue_version        = optional(string, "3.0")
    execution_class     = optional(string, "STANDARD")
    max_retries         = optional(number, 3)
    timeout             = optional(number, 2880)
    worker_type         = optional(string, "G.1X")
    number_of_workers   = optional(number, 10)
    max_concurrent_runs = optional(number, 1)
    script_object_key   = string
    default_arguments   = map(string)
  }))
}

variable "job_role_arn" {
  description = "Job Role ARN"
  type        = string
}

variable "job_tmp_bucket_name" {
  description = "Job Temporary Bucket Name"
  type        = string
}

最後に

TerraformでGlue Workflowsを実装している例がほとんど世になかったので、今回の記事ではS3 To S3のETLパイプラインの実装を公開してみました
AWS Glueは、年々サービスとしての守備範囲が広がるなど注目度が増してきているので、今後も継続的にウォッチしていきたいと思います!

参考

Discussion