[AWS Glue]パーティション化されていないパス配下のS3オブジェクトからデータを取得できるか確認してみた

[AWS Glue]パーティション化されていないパス配下のS3オブジェクトからデータを取得できるか確認してみた

#AWS Glue
#Amazon Athena
#Amazon S3
#AWS
若槻龍太

若槻龍太

facebook logohatena logotwitter logo
Clock Icon2021.01.13

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

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

AWS Glueでは、パーティション分割を行うことによりデータの整理や効率的なクエリ実行を行うことが可能です。

今回は、パーティション化されていないパス配下のS3オブジェクトからデータを取得できるか確認してみました。

確認方法

  • S3バケット上にHive形式のパスを持つオブジェクトraw-data/year=XXXX/month=XX/day=XX/object.jsonを作成
  • S3バケットのraw-dataをLocationとし、パーティションを指定しないGlueテーブルを作成
  • Glueテーブルからデータを以下の2通りの方法で取得してみる
    • AthenaからSELECTクエリで取得
    • Glueジョブからcreate_dynamic_frame.from_catalog()で取得

やってみた

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

% 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
% ETL_GLUE_JOB_NAME=devices-data-etl

環境構築

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

CloudFormationスタック

CloudFormationスタックのテンプレートです。（長いため折りたたんでいます。）

クリックで展開 
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

  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
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          Location: !Sub s3://${DevicesDataAnalyticsBucket}/integrated-data
          SerdeInfo:
            Parameters:
              paths: "device_id, timestamp, state"
            SerializationLibrary: org.apache.hive.hcatalog.data.JsonSerDe

  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
                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は、S3オブジェクトのパスをあえてパーティション化しないため、パーティションスキーマの指定はしていません。

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

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

Glueジョブスクリプト

データソースから取得したデータを変更を加えずデータターゲットに書き込むだけのPySparkスクリプトとなっています。（長いため折りたたんでいます。）

クリックで展開 
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'
).toDF()
df.show()

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

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

job.commit()

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

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

データソースへのデータ作成

パーティション化されていないパスを持つデータをS3バケットに作成します。

{"device_id": "3ff9c44a", "timestamp": 1609348014, "state": true}
% aws s3 cp raw-data.json \
  ${RAW_DATA_BUCKET}/raw-data/year=2021/month=01/day=13/raw-data.json

念の為確認してみると、データカタログにパーティションは作成されていない状態となっています。

% aws glue get-partitions \
  --database-name ${GLUE_DATABASE_NAME} \
  --table-name ${RAW_DATA_GLUE_TABLE_NAME}

{
    "Partitions": []
}

データの取得

Athenaクエリによる取得

AthenaでクエリSELECT * FROM ${RAW_DATA_GLUE_TABLE_NAME}を実行して、先程アップロードしたオブジェクトのデータを取得してみます。

% QueryExecutionId=$( \
  aws athena start-query-execution \
    --query-string "SELECT * FROM ${RAW_DATA_GLUE_TABLE_NAME}" \
    --work-group ${ATHENA_WORK_GROUP_NAME} \
    --query-execution-context Database=${GLUE_DATABASE_NAME},Catalog=AwsDataCatalog \
    --query QueryExecutionId \
    --output text \
)

クエリの実行結果を取得すると、データが取得できています。ただし、year、month、dayはパーティションスキーマとしては認識されていないようです。（Glueテーブルで定義していないので当然ですね）

% aws athena get-query-results --query-execution-id $QueryExecutionId

{
    "ResultSet": {
        "Rows": [
            {
                "Data": [
                    {
                        "VarCharValue": "device_id"
                    },
                    {
                        "VarCharValue": "timestamp"
                    },
                    {
                        "VarCharValue": "state"
                    }
                ]
            },
            {
            {
                "Data": [
                    {
                        "VarCharValue": "3ff9c44a"
                    },
                    {
                        "VarCharValue": "1609348014"
                    },
                    {
                        "VarCharValue": "true"
                    }
                ]
            }
        ],
        "ResultSetMetadata": {
            "ColumnInfo": [
                {
                    "CatalogName": "hive",
                    "SchemaName": "",
                    "TableName": "",
                    "Name": "device_id",
                    "Label": "device_id",
                    "Type": "varchar",
                    "Precision": 2147483647,
                    "Scale": 0,
                    "Nullable": "UNKNOWN",
                    "CaseSensitive": true
                },
                {
                    "CatalogName": "hive",
                    "SchemaName": "",
                    "TableName": "",
                    "Name": "timestamp",
                    "Label": "timestamp",
                    "Type": "bigint",
                    "Precision": 19,
                    "Scale": 0,
                    "Nullable": "UNKNOWN",
                    "CaseSensitive": false
                },
                {
                    "CatalogName": "hive",
                    "SchemaName": "",
                    "TableName": "",
                    "Name": "state",
                    "Label": "state",
                    "Type": "boolean",
                    "Precision": 0,
                    "Scale": 0,
                    "Nullable": "UNKNOWN",
                    "CaseSensitive": false
                }
            ]
        }
    },
    "UpdateCount": 0
}

Glueジョブによる取得

ジョブのスクリプト中でcreate_dynamic_frame.from_catalog()によるデータソースからの取得データをshow()で出力するようにしています。ジョブを実行してその出力を確認してみます。

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

ジョブのCloudWatch Logsへの出力結果を取得すると、DataFrameのデータが0件なので、データソースからデータが取得できていないでようです。

% aws logs get-log-events \
  --log-group-name /aws-glue/jobs/output \
  --log-stream-name ${JobRunId} \
  --query "events[].[message]" \
  --output text

++
||
++
++

まとめ

パーティション化されていないパス配下のS3オブジェクトのデータは、

  • Athenaからのクエリ実行であれば取得可能
    • ただし、パスをHive形式としていてもパーティションスキーマとして値は取得されない
  • Glueジョブ（create_dynamic_frame.from_catalog()）では取得不可

となりました。要するにオブジェクトがフォルダ配下にある場合のデータ取得はAthenaはOK、GlueジョブはNGということですね。

おわりに

パーティション化されていないパス配下のS3オブジェクトからデータを取得できるか確認してみました。

GlueやAthenaを触り始めた頃はこの仕様にとても混乱させられたのでちゃんと区別することを意識したいですね。

参考

以上

Share this article

facebook logohatena logotwitter logo

関連記事

AWS Glue のAmazon Q データ統合を試してみました

AWS Glue のAmazon Q データ統合を試してみました

User avatar
石川覚
2024.11.16
Amazon EventBridge SchedulerとAWS Glue Python Shellの定期実行をCDKで構築する

Amazon EventBridge SchedulerとAWS Glue Python Shellの定期実行をCDKで構築する

User avatar
niino
2024.11.07
クラスメソッド データアナリティクス通信(AWSデータ分析編) – 2024年11月号

クラスメソッド データアナリティクス通信(AWSデータ分析編) – 2024年11月号

User avatar
石川覚
2024.11.06
データ分析ツールにデータを取り込まずに検索！ - Splunk Federated Search for S3 を使ってみた

データ分析ツールにデータを取り込まずに検索！ - Splunk Federated Search for S3 を使ってみた

User avatar
酒井剛
2024.10.28
Classmethod's white logo
Facebook's logo
X's logo
YouTube's logo

© Classmethod, Inc. All rights reserved.