[AWS Glue]ジョブブックマークとPushdown Predicates併用時の動作を確認してみた

若槻龍太

2021.01.17

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、CX事業本部の若槻です。

AWS Glueでは、ジョブでデータソースからデータを取得する際にフィルターを行う方法として以下のようなものがあります。

ジョブブックマーク
Pushdown Predicates

ジョブブックマークは、以前のジョブ実行時に処理したデータを追跡し、次回以降のジョブ実行で再処理をしないようにすることが出来る機能で、データソースからのデータ取得時にも使用できます。

Pushdown Predicatesは、ジョブでデータソースからのデータ取得時にパーティションキーによるフィルターを行える機能です。

今回は、GlueジョブでこれらジョブブックマークとPushdown Predicates併用時のデータソースからのデータ取得の動作を確認してみました。

やってみた

まず最初に以降のコマンド実行で使用する変数を定義しておきます。

% AWS_REGION=ap-northeast-1
% ACCOUNT_ID=$(aws sts get-caller-identity | jq -r ".Account")
% RAW_DATA_BUCKET=s3://devices-raw-data-${ACCOUNT_ID}-${AWS_REGION}
% DATA_ANALYTICS_BUCKET=s3://devices-data-analytics-${ACCOUNT_ID}-${AWS_REGION}
% GLUE_DATABASE_NAME=devices_data_analystics
% RAW_DATA_GLUE_TABLE_NAME=devices_raw_data
% ATHENA_WORK_GROUP_NAME=devices-data-analytics

環境構築

動作確認環境を作成します。

CloudFormationスタック

CloudFormationスタックのテンプレートです。

template.yaml

AWSTemplateFormatVersion: '2010-09-09'

Resources:
  DevicesRawDataBucket:
    Type: AWS::S3::Bucket
    Properties: 
      BucketName: !Sub devices-raw-data-${AWS::AccountId}-${AWS::Region}

  DevicesDataAnalyticsAthenaWorkGroup:
    Type: AWS::Athena::WorkGroup
    Properties:
      Name: devices-data-analytics
      WorkGroupConfiguration:
        ResultConfiguration:
          OutputLocation: !Sub s3://${DevicesRawDataBucket}/query-result
        EnforceWorkGroupConfiguration: true
        PublishCloudWatchMetricsEnabled: true

  DevicesDataAnalyticsBucket:
    Type: AWS::S3::Bucket
    Properties: 
      BucketName: !Sub devices-data-analytics-${AWS::AccountId}-${AWS::Region}

  DevicesDataAnalyticsGlueDatabase:
    Type: AWS::Glue::Database
    Properties: 
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Name: devices_data_analystics

  DevicesRawDataGlueTable:
    Type: AWS::Glue::Table
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseName: !Ref DevicesDataAnalyticsGlueDatabase
      TableInput:
        Name: devices_raw_data
        TableType: EXTERNAL_TABLE
        Parameters:
          has_encrypted_data: false
          serialization.encoding: utf-8
          EXTERNAL: true
        StorageDescriptor:
          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
          Columns:
            - Name: device_id
              Type: string
            - Name: timestamp
              Type: bigint
            - Name: state
              Type: boolean
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          Location: !Sub s3://${DevicesRawDataBucket}/raw-data
          SerdeInfo:
            Parameters:
              paths: "device_id, timestamp, state"
            SerializationLibrary: org.apache.hive.hcatalog.data.JsonSerDe
        PartitionKeys:
          - Name: year
            Type: string
          - Name: month
            Type: string
          - Name: day
            Type: string

  DevicesIntegratedDataGlueTable:
    Type: AWS::Glue::Table
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseName: !Ref DevicesDataAnalyticsGlueDatabase
      TableInput:
        Name: devices_integrated_data
        TableType: EXTERNAL_TABLE
        Parameters:
          has_encrypted_data: false
          serialization.encoding: utf-8
          EXTERNAL: true
        StorageDescriptor:
          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
          Columns:
            - Name: device_id
              Type: string
            - Name: timestamp
              Type: bigint
            - Name: state
              Type: boolean
            - Name: partition_date
              Type: string
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          Location: !Sub s3://${DevicesDataAnalyticsBucket}/integrated-data
          SerdeInfo:
            Parameters:
              paths: "device_id, timestamp, state, partition_date"
            SerializationLibrary: org.apache.hive.hcatalog.data.JsonSerDe
        PartitionKeys:
          - Name: year
            Type: string
          - Name: month
            Type: string
          - Name: day
            Type: string

  ExecuteDevicesDataETLGlueJobRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          -
            Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - sts:AssumeRole
      Policies:
        - PolicyName: devices-data-etl-glue-job-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - glue:StartJobRun
                Resource:
                 - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:job/devices-data-etl
              -
                Effect: Allow
                Action:
                  - glue:GetPartition
                  - glue:GetPartitions
                  - glue:GetTable
                  - glue:BatchCreatePartition
                Resource:
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${DevicesDataAnalyticsGlueDatabase}
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${DevicesDataAnalyticsGlueDatabase}/${DevicesRawDataGlueTable}
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${DevicesDataAnalyticsGlueDatabase}/${DevicesIntegratedDataGlueTable}
              -
                Effect: Allow
                Action:
                  - s3:ListBucket
                  - s3:GetBucketLocation
                Resource: 
                  - arn:aws:s3:::*
              -
                Effect: Allow
                Action:
                  - logs:CreateLogStream
                  - logs:CreateLogGroup
                  - logs:PutLogEvents
                Resource: !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws-glue/jobs/*
              -
                Effect: Allow
                Action:
                  - s3:GetObject
                Resource:
                  - !Sub arn:aws:s3:::${DevicesRawDataBucket}/raw-data/*
                  - !Sub arn:aws:s3:::${DevicesDataAnalyticsBucket}/glue-job-script/devices-data-etl.py
              -
                Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                Resource:
                  - !Sub arn:aws:s3:::${DevicesDataAnalyticsBucket}/glue-job-temp-dir/*
              -
                Effect: Allow
                Action:
                  - s3:PutObject
                Resource:
                  - !Sub arn:aws:s3:::${DevicesDataAnalyticsBucket}/integrated-data/*

  DevicesDataETLGlueJob:
    Type: AWS::Glue::Job
    Properties:
      Name: devices-data-etl
      Command:
        Name: glueetl
        PythonVersion: 3
        ScriptLocation: !Sub s3://${DevicesDataAnalyticsBucket}/glue-job-script/devices-data-etl.py
      DefaultArguments:
        --job-language: python
        --job-bookmark-option: job-bookmark-enable
        --TempDir: !Sub s3://${DevicesDataAnalyticsBucket}/glue-job-temp-dir
        --GLUE_DATABASE_NAME: !Sub ${DevicesDataAnalyticsGlueDatabase}
        --RAW_DATA_GLUE_TABLE_NAME: !Sub ${DevicesRawDataGlueTable}
        --INTEGRATED_DATA_GLUE_TABLE_NAME: !Sub ${DevicesIntegratedDataGlueTable}
      GlueVersion: 2.0
      ExecutionProperty:
        MaxConcurrentRuns: 1
      MaxRetries: 0
      Role: !Ref ExecuteDevicesDataETLGlueJobRole

データソースとなるGlueテーブルのリソース定義DevicesRawDataGlueTableで、Properties.TableInput.PartitionKeysにて、year、month、dayの3つのパーティションキーを定義しています。

ジョブのリソース定義DevicesDataETLGlueJobで、Properties.DefaultArgumentsにて引数job-bookmark-optionにjob-bookmark-enableを指定してジョブブックマークを有効化しています。

スタックをデプロイします。

% aws cloudformation deploy \
  --template-file template.yaml \
  --stack-name devices-data-analytics-stack \
  --capabilities CAPABILITY_NAMED_IAM \
  --no-fail-on-empty-changeset

Glueジョブスクリプト

データソースから取得したデータをパーティション情報を維持したままデータターゲットに書き込むだけのPySparkスクリプトです。

devices-data-etl.py

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame

args = getResolvedOptions(
  sys.argv,
  [
    'JOB_NAME',
    'GLUE_DATABASE_NAME',
    'RAW_DATA_GLUE_TABLE_NAME',
    'INTEGRATED_DATA_GLUE_TABLE_NAME'
  ]
)

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

df = glueContext.create_dynamic_frame.from_catalog(
  database = args['GLUE_DATABASE_NAME'],
  table_name = args['RAW_DATA_GLUE_TABLE_NAME'],
  transformation_ctx = 'datasource',
  push_down_predicate = "year='2021' and month='01' and day='17'"
).toDF()
df.show()

dyf = DynamicFrame.fromDF(
  df,
  glueContext,
  'integrated_data'
)

additionalOptions = {
  "enableUpdateCatalog": True
}

additionalOptions['partitionKeys'] = [
  'year', 'month', 'day'
]

glueContext.write_dynamic_frame.from_catalog(
  frame = dyf,
  database = args['GLUE_DATABASE_NAME'],
  table_name = args['INTEGRATED_DATA_GLUE_TABLE_NAME'],
  transformation_ctx = 'datasink',
  additional_options=additionalOptions
)

job.commit()

create_dynamic_frame.from_catalog()でpush_down_predicateオプションを指定して、データソースから取得するデータをPushdown Predicatesによりパーティションキーyear=2021/month=01/day=17でフィルターするようにしています。

スクリプトをS3バケットにアップロードします。

% aws s3 cp devices-data-etl.py \
  ${DATA_ANALYTICS_BUCKET}/glue-job-script/devices-data-etl.py

動作確認

1回目のジョブ実行：Pushdown Predicatesの動作確認

まず、今回作成したジョブのPushdown Predicatesの対象とならないyear=2021/month=01/day=16と、対象となるyear=2021/month=01/day=17のパーティションパスを持つオブジェクトを、データソースのS3ロケーションにそれぞれ作成します。

raw-data-1.json

{"device_id": "device_1", "timestamp": 1609348014, "state": true}

% aws s3 cp raw-data-1.json \
  ${RAW_DATA_BUCKET}/raw-data/year=2021/month=01/day=16/raw-data-1.json
% aws s3 cp raw-data-1.json \
  ${RAW_DATA_BUCKET}/raw-data/year=2021/month=01/day=17/raw-data-1.json

AthenaでMSCK REPAIR TABLEクエリを実行してアップロードしたオブジェクトのパーティションキーに対応するパーティションを作成します。

% aws athena start-query-execution \
  --query-string "MSCK REPAIR TABLE ${GLUE_TABLE_NAME}" \
  --work-group ${ATHENA_WORK_GROUP_NAME} \
  --query-execution-context \
    Database=${GLUE_DATABASE_NAME},Catalog=AwsDataCatalog

ジョブを実行します。

% JobRunId=$(
  aws glue start-job-run --job-name ${ETL_GLUE_JOB_NAME} \
    --query JobRunId \
    --output text
)

ジョブのスクリプト中でデータソースからの取得データをdf.show()で出力するようにしています。CloudWatch Logsへの出力結果を取得すると、Pushdown Predicatesによりパーティションがyear=2021/month=01/day=17のデータのみ取得できていること確認できます。

% aws logs get-log-events \
  --log-group-name /aws-glue/jobs/output \
  --log-stream-name ${JobRunId} \
  --query "events[].[message]" \
  --output text
+---------+----------+-----+----+-----+---+
|device_id| timestamp|state|year|month|day|
+---------+----------+-----+----+-----+---+
| device_1|1609348014| true|2021|   01| 17|
+---------+----------+-----+----+-----+---+

2回目のジョブ実行：ジョブブックマークとPushdown Predicates併用時の動作確認

year=2021/month=01/day=17のパーティションパスを持つオブジェクトを、データソースのS3ロケーションに作成します。

raw-data-2.json

{"device_id": "device_2", "timestamp": 1609348014, "state": true}

% aws s3 cp raw-data-2.json \
  ${RAW_DATA_BUCKET}/raw-data/year=2021/month=01/day=17/raw-data-2.json

これでパーティションパスyear=2021/month=01/day=17配下に1回目に処理済みのオブジェクトと未処理のオブジェクトが1つずつ作成されている状態となっています。

% aws s3 ls ${RAW_DATA_BUCKET}/raw-data/year=2021/month=01/day=17/ --recursive
2021-01-17 04:26:11         65 raw-data/year=2021/month=01/day=17/raw-data-1.json
2021-01-17 05:20:34         65 raw-data/year=2021/month=01/day=17/raw-data-2.json

ジョブを実行します。

% JobRunId=$(
  aws glue start-job-run --job-name ${ETL_GLUE_JOB_NAME} \
    --query JobRunId \
    --output text
)

CloudWatch Logsへの出力結果を取得すると、Pushdown Predicatesによりパーティションがyear=2021/month=01/day=17、かつジョブブックマークにより1回目で未処理のオブジェクトに記載されたデータのみ取得できていることが確認できます。

% aws logs get-log-events \
  --log-group-name /aws-glue/jobs/output \
  --log-stream-name ${JobRunId} \
  --query "events[].[message]" \
  --output text
+---------+----------+-----+----+-----+---+
|device_id| timestamp|state|year|month|day|
+---------+----------+-----+----+-----+---+
| device_2|1609348014| true|2021|   01| 17|
+---------+----------+-----+----+-----+---+

まとめ

動作確認の結果、AWS Glueジョブでデータソースからのデータ取得時に「ジョブブックマーク」と「Pushdown Predicates」が併用された時は、いずれも作用して取得が行われるという結果となりました。

予想通りの結果でしたがちゃんと確認ができて良かったです。

参考

[AWS Glue]データカタログのパーティションをAWS CLIで更新してみた | Developers.IO

以上

[AWS Glue]ジョブブックマークとPushdown Predicates併用時の動作を確認してみた

やってみた

環境構築

CloudFormationスタック

Glueジョブスクリプト

動作確認

1回目のジョブ実行：Pushdown Predicatesの動作確認

2回目のジョブ実行：ジョブブックマークとPushdown Predicates併用時の動作確認

まとめ

参考

イベント

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS / システムアーキテクト案件特集〜を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩

[AWS Glue]ジョブブックマークとPushdown Predicates併用時の動作を確認してみた

やってみた

環境構築

CloudFormationスタック

Glueジョブスクリプト

動作確認

1回目のジョブ実行：Pushdown Predicatesの動作確認

2回目のジョブ実行：ジョブブックマークとPushdown Predicates併用時の動作確認

まとめ

参考

イベント

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会 〜AWS / システムアーキテクト案件特集〜 を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金） 名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制 はじめの一歩

関連記事

Glueクローラーのデータソースとして別AWSアカウントのS3バケットを指定したい

GitHub Actionsでコンテナ展開時にno space left on deviceで失敗する場合の暫定対応

Lambda+Glue+Step Functionsの構成をServerless FrameworkとAWS SAMのそれぞれでデプロイしてみた

AWS Glue crawler のクロールが終わったら Lambda を実行する

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS / システムアーキテクト案件特集〜を開催します

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩