Amazon CloudFrontのアクセスログ保存パスをAthena用に最適化してみた~part2~

はじめに

こんにちは。2年目の大林です。 本ブログでは前回のブログ(Amazon CloudFrontのアクセスログ保存パスをAthena用に最適化してみた~part1~)の続きを書いていきたいと思います。 前回のブログでは、S3バケットに新規でアップロードされたログファイルに対してパス最適化処理を行う方法を紹介しました。 今回は、既にS3バケットにアップロードされているログファイルに対してパス最適化処理を行う方法を紹介したいと思います。

想定しているケース

ケース1
　・CloudFrontのアクセスログが有効になっていて、既にS3バケットにログファイルが保存されている場合
　・単一のS3バケットを使用したパス最適化を実施したい
ケース2
　・CloudFrontのアクセスログが有効になっていて、既にS3バケットにログファイルが保存されている場合
　・パス最適化処理の途中でエラーが起きた場合でもログファイルを消失したくない場合

ケース1

ケース1の設計と処理の流れ

①Amazon CloudFrontのアクセスログがS3バケットに保存されている

②Glueを使用して、アクセスログ保存のパスを最適化する

・ログファイルをパス（ディストリビューション名/yyyy/mm/dd）を指定してコピーします。
・上記の処理が終了した時点で、パス最適化前と最適化後のログファイルが存在することになるのでパスが最適化できていないほうを削除します。

使用するスクリプトは以下です。

import re
import boto3
import asyncio
import os

s3 = boto3.client('s3')
date_pattern = r'[^\\d](\d{4})-(\d{2})-(\d{2})-(\d{2})[^\\d]'
filename_pattern = r'[^/]+$'
bucket = ログファイルが格納されているS3バケット
target = '/'
client = boto3.client('s3')
obj = client.list_objects(Bucket=bucket)
obj_list = obj['ResponseMetadata']
data = obj['Contents']
keys = [item['Key'] for item in data if not item['Key'].endswith('/')]

for source_key in keys:
    target_find = source_key.find(target)
    target_key_prefix = source_key[:target_find]
    source_regex = re.compile(date_pattern)
    match = source_regex.search(source_key)

    if match is None:
        print('ファイルが見つかりませんでした。')
    else:
        year, month, day, hour = match.groups()
        filename_regex = re.compile(filename_pattern)
        filename = filename_regex.search(source_key).group(0)
        target_key = target_key_prefix + '/' + year + '/' + month + '/' + day + '/' + filename
        print(str(source_key) + '  to  ' + str(target_key))
        copy_result =s3.copy_object(Bucket=bucket, Key=target_key, 
          CopySource={'Bucket': bucket, 'Key': source_key})
        print(copy_result)
        delete_result = s3.delete_object(Bucket=bucket, Key=source_key)
        print(delete_result)

③Amazon Athenaでアクセスログをクエリする

分析用のテンプレートは以下です。

AWSTemplateFormatVersion: "2010-09-09"
#----------------------------------------------------------------------#
# Athenaを利用したCloudFrontのアクセスログ分析用テンプレート
# 作成されるリソース：Glue:Database, Glue:Table, S3bucket, Athenaワークグループ
#----------------------------------------------------------------------#
Parameters:
 #アカウントIDを入力
  AccountId:
    Description: "input Account Id "
    Type: String
 #環境名を入力
  Env:
    Description: dev or prod
    Type: String
 #システム名を入力
  SystemName:
    Description: System name
    Type: String
 #CloudFrontのアクセスログが保存されているS3バケットを指定する
  S3bucketCloudFrontLogs:
    Type: String
 #オブジェクトの有効期限を入力
  ExpirationInDays:
    Type: String
    Default: 1825
 #Glueテーブルのパーティションに設定するCloudFrontディストリビューションを入力
  distributions:
    Description: "enter distribution name by Comma Delimiter list (XXXXXXXXXXXX,XXXXXXXXXXXX...)"
    Type: String

Resources:
#------------------------------------------------------------#
# Glue
# アクセスログ用のデータベースとテーブルを作成する
#------------------------------------------------------------#
  GlueDatabase:
    Type: AWS::Glue::Database
    Properties: 
      CatalogId: !Ref AWS::AccountId  
      DatabaseInput:
        Name: !Sub '${Env}_${SystemName}_cloudfront_database'
   
  #CloudFrontアクセスログのGlueテーブル   
  GlueTableCloudFrontLogs:
    Type: AWS::Glue::Table
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseName: !Ref GlueDatabase
      TableInput:
        Name: !Sub '${Env}_${SystemName}_cloudfront_table'
        TableType: EXTERNAL_TABLE
        Parameters:
          EXTERNAL: true
          projection.enabled: true
          projection.distribution.type: enum
          projection.distribution.values: !Sub '${distributions}'
          projection.datetime.type: date
          projection.datetime.format: yyyy/MM/dd
          projection.datetime.range: 2023/01/01,NOW
          storage.location.template: !Sub 's3://${S3bucketCloudFrontLogs}/${!distribution}/${!datetime}'
          skip.header.line.count: '2'
          serialization.encoding: utf-8
        PartitionKeys:
          - Name: distribution
            Type: string
          - Name: datetime
            Type: string
        StorageDescriptor:
          Columns:
            - Name: date
              Type: date
            - Name: time
              Type: string
            - Name: location
              Type: string
            - Name: bytes
              Type: bigint
            - Name: requestip
              Type: string
            - Name: method
              Type: string
            - Name: host
              Type: string
            - Name: uri
              Type: string
            - Name: status
              Type: int
            - Name: referrer
              Type: string
            - Name: useragent
              Type: string
            - Name: querystring
              Type: string
            - Name: cookie
              Type: string
            - Name: resulttype
              Type: string
            - Name: requestid
              Type: string
            - Name: hostheader
              Type: string
            - Name: requestprotocol
              Type: string
            - Name: requestbytes
              Type: bigint
            - Name: timetaken
              Type: float
            - Name: xforwardedfor
              Type: string
            - Name: sslprotocol
              Type: string
            - Name: sslcipher
              Type: string
            - Name: responseresulttype
              Type: string
            - Name: httpversion
              Type: string
            - Name: filestatus
              Type: string
            - Name: encryptedfields
              Type: int
            - Name: cPort
              Type: string
            - Name: timeToFirstByte
              Type: string       
            - Name: xEdgeDetailedResultType
              Type: string
            - Name: scContentType
              Type: string   
            - Name: scContentLen
              Type: string 
            - Name: scRangeStart
              Type: string
            - Name: scRangeEnd
              Type: string                   
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
          Location: !Sub 's3://${S3bucketCloudFrontLogs}/'
          SerdeInfo:
            SerializationLibrary: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
            Parameters:
              serialization.format: "\t"
  #------------------------------------------------------------#
  # Athena
  #------------------------------------------------------------#
  AthenaWorkGroup:
    Type: AWS::Athena::WorkGroup
    Properties:
      Name: !Sub "${Env}-athena-work-group"
      WorkGroupConfiguration:
        ResultConfiguration:
          OutputLocation: !Sub "s3://${AthenaQueryResultBucket}/data"
        EnforceWorkGroupConfiguration: true
        PublishCloudWatchMetricsEnabled: true
  #------------------------------------------------------------#
  # S3 Backet
  # Athenaでクエリを実行した結果が格納されるS3バケットを作成する
  #------------------------------------------------------------#
  AthenaQueryResultBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: 'Retain'
    Properties: 
      BucketName:  !Sub '${Env}-${SystemName}-cloudfront-athena-result'
      AccessControl: Private
      PublicAccessBlockConfiguration:
        BlockPublicAcls: True
        BlockPublicPolicy: True
        IgnorePublicAcls: True
        RestrictPublicBuckets: True
      LifecycleConfiguration:
        Rules:
          - Id: life-cycle-rule
            Status: Enabled
            ExpirationInDays: !Sub '${ExpirationInDays}'
      BucketEncryption:
        ServerSideEncryptionConfiguration: 
          -
            ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256

ケース1のデモ

添付画像の通り、すでに指定したプレフィックス名のフォルダ配下にログがS3バケットにアップロードされています。
上記のスクリプトの9行目で処理対象のS3バケットを指定して、ジョブを実行します。 実行結果は以下の添付画像の通りです。
意図した通り、「バケット名/ディストリビューション名/yyyy/mm/dd」というパスにログファイルが保存されています。

ケース2

ケース2の設計と処理の流れ

ケース2では、パス最適化処理中にエラーが発生した場合でもログファイルが消失してしまうリスクを抑えた構成になっています。
一方で、S3バケットを2つ使用するためその分コストがかかってしまいます。 そのため、ライフサイクルポリシーを使用するなどコストを抑える対策を取る必要があります。

①Amazon CloudFrontのアクセスログがS3バケットに保存されている

②Glueを使用して、アクセスログ保存のパスを最適化する

・ログファイルをパス（ディストリビューション名/yyyy/mm/dd）を指定してコピーします。
・上記の処理が終了した時点で、パス最適化前と最適化後のログファイルが存在することになるのでパスが最適化できていないほうを削除します。
パス（ディストリビューション名/yyyy/mm/dd）を指定してログファイルを別のS3バケットにコピーします。
使用するスクリプトは以下です。

import re
import boto3
import asyncio
import os

s3 = boto3.client('s3')
date_pattern = r'[^\\d](\d{4})-(\d{2})-(\d{2})-(\d{2})[^\\d]'
filename_pattern = r'[^/]+$'
bucket = ログファイルが格納されているS3バケット
destination_bucket = ログファイルのコピー先S3バケット
target = '/'
client = boto3.client('s3')
obj = client.list_objects(Bucket='y-obayashi-0910')
obj_list = obj['ResponseMetadata']
data = obj['Contents']
keys = [item['Key'] for item in data if not item['Key'].endswith('/')]

for source_key in keys:
    target_find = source_key.find(target)
    target_key_prefix = source_key[:target_find]
    source_regex = re.compile(date_pattern)
    match = source_regex.search(source_key)
    if match is None:
        print('ファイルが見つかりませんでした。')
    else:
        year, month, day, hour = match.groups()
        filename_regex = re.compile(filename_pattern)
        filename = filename_regex.search(source_key).group(0)
        target_key = target_key_prefix + '/' + year + '/' + month + '/' + day + '/' + filename
        print(str(source_key) + '  to  ' + str(target_key))
        copy_result =s3.copy_object(Bucket=destination_bucket, Key=target_key, 
          CopySource={'Bucket': bucket, 'Key': source_key})
        print(copy_result)

③Amazon Athenaでアクセスログをクエリする

分析用のテンプレートはケース1と同様のものを使用します。

ケース2のデモ

添付画像の通り、すでに指定したプレフィックス名のフォルダ配下にログがS3バケットにアップロードされています。
上記のスクリプトの9行目でログファイルが格納されているS3バケット、10行目でログファイルのコピー先S3バケット指定して、ジョブを実行します。 実行結果は以下の添付画像の通りです。
意図した通り、別のS3バケットに「バケット名/ディストリビューション名/yyyy/mm/dd」というパスにログファイルが保存されています。

さいごに

今回は、既にS3バケットにアップロードされているログファイルに対してパス最適化処理を行う方法を紹介しました。
本ブログでは「ディストリビューション名/yyyy/mm/dd」といったパスでログファイルを保存するようにしましたが、どういったパスにログファイルを保存するかはケースによって異なってくると思うので状況を見て判断する必要があると考えています。
この検証を通して、細かくパーティションを切っていけばいいという単純なものではないことを学びました。
パーティション設定…奥が深いです…。

執筆者大林 優斗

クラウドエンジニア
AWSを活用したシステムの設計と開発をやらせていただいています。

執筆記事一覧：https://tech.nri-net.com/archive/author/y-obayashi