如何使用分区的 Amazon S3 访问日志防止 Athena 查询超时？

解决方法

Amazon S3 访问日志使用相同的前缀存储。如果有大量数据，则 Athena 查询可能会在 Athena 读取所有数据之前超时。为防止出现此问题，请使用 AWS Glue ETL 作业对您的 Amazon S3 数据进行分区。然后，在有限的分区上运行 Athena 查询。

**注意：**对于示例表、脚本和命令，必要时将以下值替换为您的值：

• 将 s3_access_logs_db 替换为您的数据库名称
• 将 s3://awsexamplebucket1-logs/prefix/ 替换为存储您的 Amazon S3 访问日志的路径
• 将 s3_access_logs 替换为您的表名
• 将 s3_access_logs_partitioned 替换为您的分区表名称
• 将 2023、03 和 04 替换为您的分区值

对 Amazon S3 数据进行分区

在 Athena 中创建下表：

CREATE EXTERNAL TABLE `s3_access_logs_db.s3_access_logs`(    
  `bucketowner` string,
  `bucket_name` string,
  `requestdatetime` string,
  `remoteip` string,
  `requester` string,
  `requestid` string,
  `operation` string,
  `key` string,
  `request_uri` string,
  `httpstatus` string,
  `errorcode` string,
  `bytessent` string,
  `objectsize` string,
  `totaltime` string,
  `turnaround_time` string,
  `referrer` string,
  `useragent` string,
  `version_id` string,
  `hostid` string,
  `sigv` string,
  `ciphersuite` string,
  `authtype` string,
  `endpoint` string,
  `tlsversion` string,
  `accesspoint_arn` string,
  `aclrequired` string)
ROW FORMAT SERDE
  'com.amazonaws.glue.serde.GrokSerDe'
WITH SERDEPROPERTIES (
  'input.format'='%{NOTSPACE:bucketowner} %{NOTSPACE:bucket_name} \\[%{INSIDE_BRACKETS:requestdatetime}\\] %{NOTSPACE:remoteip} %{NOTSPACE:requester} %{NOTSPACE:requestid} %{NOTSPACE:operation} %{NOTSPACE:key} \"%{INSIDE_QS:request_uri}\" %{NOTSPACE:httpstatus} %{NOTSPACE:errorcode} %{NOTSPACE:bytes_sent} %{NOTSPACE:objectsize} %{NOTSPACE:totaltime} %{NOTSPACE:turnaround_time} \"?%{INSIDE_QS:referrer}\"? \"%{INSIDE_QS:useragent}\" %{NOTSPACE:version_id} %{NOTSPACE:hostid} %{NOTSPACE:sigv} %{NOTSPACE:ciphersuite} %{NOTSPACE:authtype} %{NOTSPACE:endpoint} %{NOTSPACE:tlsversion}( %{NOTSPACE:accesspoint_arn} %{NOTSPACE:aclrequired})?',
  'input.grokCustomPatterns'='INSIDE_QS ([^\"]*)\nINSIDE_BRACKETS ([^\\]]*)')
STORED AS INPUTFORMAT
  'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://awsexamplebucket1-logs/prefix/';

创建 AWS Glue ETL 作业

完成以下步骤：

1. 打开 AWS Glue 控制台。

2. 选择 ETL jobs（ETL 作业），然后选择 Spark script editor（Spark 脚本编辑器）。

3. 选择 Create（创建）。

4. 在脚本选项卡上，输入以下脚本：

   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 pyspark.sql.functions import split, col, size
   from awsglue.dynamicframe import DynamicFrame
   
   ## @params: [JOB_NAME]
   args = getResolvedOptions(sys.argv, ['JOB_NAME'])
   
   sc = SparkContext()
   glueContext = GlueContext(sc)
   spark = glueContext.spark_session
   job = Job(glueContext)
   job.init(args['JOB_NAME'], args)
   
   dyf = glueContext.create_dynamic_frame.from_catalog(database='s3_access_logs_db', table_name='s3_access_logs', transformation_ctx = 'dyf',additional_options = {"attachFilename": "s3path"})
   
   df = dyf.toDF()
   df2=df.withColumn('filename',split(col("s3path"),"/"))\
           .withColumn('year',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(0))\
           .withColumn('month',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(1))\
           .withColumn('day',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(2))\
           .drop('s3path','filename')
   
   output_dyf = DynamicFrame.fromDF(df2, glue_ctx=glueContext, name = 'output_dyf')
   
   partitionKeys = ['year', 'month', 'day']
   sink = glueContext.getSink(connection_type="s3", path='s3://awsexamplebucket2-logs/prefix/',
                              enableUpdateCatalog=True, updateBehavior="UPDATE_IN_DATABASE",
                              partitionKeys=partitionKeys)
   sink.setFormat("glueparquet")
   sink.setCatalogInfo(catalogDatabase='s3_access_logs_db', catalogTableName='s3_access_logs_partitioned')
   sink.writeFrame(output_dyf)
   
   job.commit()

5. 在 Job details（作业详细信息）选项卡上，输入您的作业名称，然后选择 IAM role（IAM 角色）。

6. 选择 Save（保存），然后选择 Run（运行）。

**注意：**Amazon S3 访问日志会定期送达。要为 AWS Glue ETL 作业设置基于时间的计划，请添加触发器。另外，打开作业书签。

创建 DynamicFrame 对象和分区表

创建 Amazon S3 访问日志表后，创建包含 Amazon S3 访问日志的 DynamicFrame 对象。然后，您可以创建带有键的分区表，以基于 DynamicFrame 对象显示年、月和日。

完成以下步骤：

1. 运行以下命令以创建 DynamicFrame 对象并扫描 Amazon S3 访问日志表：

     dyf = glueContext.create_dynamic_frame.from_catalog(database='s3_access_logs_db, table_name='s3_access_logs', transformation_ctx = 'dyf',additional_options = {"attachFilename": "s3path"})

   注意：attachFilename 参数用作列名。

2. 运行以下命令以从 Amazon S3 访问日志的路径中创建年、月和日列：

   df = dyf.toDF()df2=df.withColumn('filename',split(col("s3path"),"/"))\
           .withColumn('year',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(0))\
           .withColumn('month',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(1))\
           .withColumn('day',split(col("filename").getItem(size(col("filename"))-1),"-").getItem(2))\
           .drop('s3path','filename')

3. 运行以下命令以为 Amazon S3 访问日志创建分区表：

   output_dyf = DynamicFrame.fromDF(df2, glue_ctx=glueContext, name = 'output_dyf')
   partitionKeys = ['year', 'month', 'day']
   sink = glueContext.getSink(connection_type="s3", path='s3://awsexamplebucket2-logs/prefix/',
                              enableUpdateCatalog=True, updateBehavior="UPDATE_IN_DATABASE",
                              partitionKeys=partitionKeys)
   sink.setFormat("glueparquet")
   sink.setCatalogInfo(catalogDatabase='s3_access_logs_db', catalogTableName='s3_access_logs_partitioned')
   sink.writeFrame(output_dyf)

查询分区表

1. 打开 Athena 控制台。
2. 运行以下命令来查询该表并确认该表已分区：

   SELECT * FROM "s3_access_logs_db"."s3_access_logs_partitioned" WHERE year = '2023' AND month = '03' AND day = '04'