概要
やりたいこと
Athena のデータスキャン量が多い。パーティションを使っているのに。Redash がすぐ死ぬ。軽くしたい。
Athena のチューニングをしようと思い下記を確認。
Parquet フォーマットが良いらしい。
Parquet フォーマットは Kinesis Firehose で取り扱えるとのこと。
AWS::KinesisFirehose::DeliveryStream の「レコード形式を変換」を参考につくる。
Athena は列指向なので Parquet がいい
Athena が扱えるフォーマットには以下がある。
この列方向にデータを分割して格納することを垂直パーティショニングと言う。さらに内部で行方向に一定の単位でメタ情報を保持することによりスキャン範囲を限定できる水平パーティショニングの機能も持っている。
詳しくは下のブログを読んでくれ！
CloudFormation で反映
AWS::KinesisFirehose::DeliveryStream を参考に AWS WAF 用に変換していく。
AWS WAF 用の Glue Table
Table 構造は AWS WAF のものにしておく。
rulegrouplist について、クラスメソッドの記事で提案されている TYPE のほうが詳細だが、自分の環境ではエラーが起きたため、公式に準拠した Table 定義にしておく。
Resources:
GlueDatabase:
Type: AWS::Glue::Database
Properties:
CatalogId: !Ref AWS::AccountId
DatabaseInput:
Name: glue-db-datacatalog
GlueTable:
Type: AWS::Glue::Table
Properties:
CatalogId: !Ref AWS::AccountId
DatabaseName: !Ref GlueDatabase
TableInput:
Name: glue-waf-datacatalog
Owner: owner
Retention: 0
StorageDescriptor:
Columns:
- Name: timestamp
Type: bigint
- Name: formatversion
Type: int
- Name: webaclid
Type: string
- Name: terminatingruleid
Type: string
- Name: terminatingruletype
Type: string
- Name: action
Type: string
- Name: terminatingrulematchdetails
Type: array<struct<conditiontype:string,location:string,matcheddata:array<string>>>
- Name: httpsourcename
Type: string
- Name: httpsourceid
Type: string
- Name: rulegrouplist
Type: array<string>
- Name: ratebasedrulelist
Type: array<struct<ratebasedruleid:string,limitkey:string,maxrateallowed:int>>
- Name: nonterminatingmatchingrules
Type: array<struct<ruleid:string,action:string>>
- Name: httprequest
Type: struct<clientIp:string,country:string,headers:array<struct<name:string,value:string>>,uri:string,args:string,httpVersion:string,httpMethod:string,requestId:string>
InputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
OutputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
Compressed: false
NumberOfBuckets: -1
SerdeInfo:
SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
Parameters:
serialization.format: '1'
BucketColumns: []
SortColumns: []
StoredAsSubDirectories: false
PartitionKeys:
- Name: year
Type: string
- Name: month
Type: string
- Name: day
Type: string
- Name: hour
Type: string
TableType: EXTERNAL_TABLE
Firehose 用の IAM Role
公式より、データ形式変換のために AWS Glue に Kinesis Data Firehose アクセス権を付与する必要がある。
-
Effect: Allow
Action:
- glue:GetTable
- glue:GetTableVersion
- glue:GetTableVersions
Resource: "*"
全体像は下記。
FirehoseRole:
Type: AWS::IAM::Role
Properties:
RoleName: FirehoseParquetRole
Path: '/'
AssumeRolePolicyDocument: # Kinesis FirehoseにIAM Roleを割り当てるため
Version: 2012-10-17
Statement:
- Effect: Allow
Principal:
Service:
- firehose.amazonaws.com
Action:
- sts:AssumeRole
Condition:
StringEquals:
sts:ExternalId: !Sub ${AWS::AccountId}
Policies:
-
PolicyName: firehose_delivery_role
PolicyDocument:
Version: "2012-10-17"
Statement:
-
Effect: Allow
Action:
- glue:GetTable
- glue:GetTableVersion
- glue:GetTableVersions
Resource: "*"
-
Effect: "Allow"
Action:
- s3:AbortMultipartUpload
- s3:GetBucketLocation
- s3:GetObject
- s3:ListBucket
- s3:ListBucketMultipartUploads
- s3:PutObject
Resource:
- !Sub arn:aws:s3:::${BucketName}
- !Sub arn:aws:s3:::${BucketName}/*
-
Effect: "Allow"
Action:
- kinesis:DescribeStream
- kinesis:GetShardIterator
- kinesis:GetRecords
- kinesis:ListShards
Resource:
- !Sub arn:aws:kinesis:${AWS::Region}:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
-
Effect: "Allow"
Action:
- kms:GenerateDataKey
- kms:Decrypt
Resource:
- !Sub arn:aws:kms:${AWS::Region}:${AWS::AccountId}:alias/aws/s3
Condition:
StringEquals:
kms:ViaService: !Sub s3.${AWS::Region}.amazonaws.com
StringLike:
kms:EncryptionContext:aws:s3:arn:
- !Sub arn:aws:s3:::${BucketName}/*
- !Sub arn:aws:s3:::${BucketName}/%FIREHOSE_BUCKET_PREFIX%*
-
Effect: "Allow"
Action:
- logs:PutLogEvents
Resource:
- !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/aws-waf-logs-${AWS::Region}-${WebACLName}:log-stream:*
-
Effect: "Allow"
Action:
- lambda:InvokeFunction
- lambda:GetFunctionConfiguration
Resource:
- !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:%FIREHOSE_DEFAULT_FUNCTION%:%FIREHOSE_DEFAULT_VERSION%
-
Effect: "Allow"
Action:
- kms:Decrypt
Resource:
- !Sub arn:aws:kms:${AWS::Region}:${AWS::AccountId}:key/%SSE_KEY_ID%
Condition:
StringEquals:
kms:ViaService: kinesis.%REGION_NAME%.amazonaws.com
StringLike:
kms:EncryptionContext:aws:kinesis:arn: !Sub arn:aws:kinesis:%REGION_NAME%:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
Kinesis Firehose
ExtendedS3DestinationConfiguration を利用し、DataFormatConversionConfiguration で OutputFormatConfiguration で Parquet を指定する。
FirehoseDeliveryStream:
Type: "AWS::KinesisFirehose::DeliveryStream"
Properties:
DeliveryStreamName: !Sub aws-waf-logs-parquet
DeliveryStreamType: DirectPut
ExtendedS3DestinationConfiguration:
RoleARN: !GetAtt FirehoseRole.Arn
BucketARN: !Sub arn:aws:s3:::${BucketName}
Prefix: !Join
- ''
- - "WAFLOG"
- '/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/'
ErrorOutputPrefix: !Join
- ''
- - "ERRORLOG"
- '/!{firehose:error-output-type}/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/'
BufferingHints:
SizeInMBs: 128
IntervalInSeconds: 300
CompressionFormat: UNCOMPRESSED
EncryptionConfiguration:
NoEncryptionConfig: NoEncryption
CloudWatchLoggingOptions:
Enabled: true
LogGroupName: !Sub '/aws/kinesisfirehose/aws-waf-logs-${AWS::StackName}'
LogStreamName: S3Delivery
DataFormatConversionConfiguration:
SchemaConfiguration:
CatalogId: !Ref AWS::AccountId
RoleARN: !GetAtt FirehoseRole.Arn
DatabaseName: !Ref GlueDatabase
TableName: glue-table-waflog #!Ref GlueTable
Region: !Ref AWS::Region
VersionId: LATEST
InputFormatConfiguration:
Deserializer:
OpenXJsonSerDe: {}
OutputFormatConfiguration:
Serializer:
ParquetSerDe: {}
Enabled: True
Kinesis Firehose の Conver record format の項目が下記のようになる。
S3 にエクスポートされたログを確認すると parquet になっている。
S3 Select からファイル形式を Parquet でファイルプレビューをすると json が表示されること。
コメント