Comment utiliser un journal d'accès Amazon S3 partitionné pour empêcher l'expiration d'une requête Athena ?

Résolution

Les journaux d'accès Amazon S3 sont stockés avec le même préfixe. S'il existe une grande quantité de données, les requêtes Athena peuvent expirer avant qu'Athena ne lise toutes les données. Pour éviter ce problème, utilisez une tâche ETL AWS Glue pour partitionner vos données Amazon S3. Puis, exécutez les requêtes Athena sur des partitions limitées.

Remarque : Pour l'exemple de table, de script et de commandes, remplacez les valeurs suivantes par les vôtres si nécessaire :

• s3_access_logs_db par le nom de votre base de données
• s3://awsexamplebucket1-logs/prefix/ par le chemin qui stocke vos journaux d'accès Amazon S3
• s3_access_logs par le nom de votre table
• s3_access_logs_partitioned par le nom de votre table partitionnée
• 2023, 03 et 04 par vos valeurs de partition

Partitionner les données Amazon S3

Créez la table suivante dans 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/';

Créer une tâche ETL AWS Glue

Procédez comme suit :

1. Ouvrez la console AWS Glue.

2. Sélectionnez Tâches ETL, puis choisissez l'éditeur de script Spark.

3. Sélectionnez Créer.

4. Dans l'onglet Script, saisissez le script suivant :

   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. Dans l'onglet Détails de la tâche, saisissez le nom de votre tâche, puis sélectionnez Rôle IAM.

6. Sélectionnez Enregistrer, puis Exécuter.

Remarque : Les journaux d'accès Amazon S3 sont régulièrement livrés. Pour définir des calendriers temporels pour la tâche ETL AWS Glue, ajoutez un déclencheur. Activez également les signets de tâche.

Créer un objet DynamicFrame et une table partitionnée

Après avoir créé la table de journaux d'accès Amazon S3, créez un objet DynamicFrame contenant les journaux d'accès Amazon S3. Vous pouvez ensuite créer une table partitionnée avec des clés pour afficher l'année, le mois et le jour en fonction de l'objet DynamicFrame.

Procédez comme suit :

1. Exécutez la commande suivante pour créer un objet DynamicFrame et analysez la table de journaux d'accès 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"})

   Remarque : Le paramètre attachFilename est utilisé comme nom de colonne.

2. Exécutez la commande suivante pour créer des colonnes d'année, de mois et de jour à partir du chemin d'accès à vos journaux d'accès 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. Exécutez la commande suivante pour créer une table partitionnée pour les journaux d'accès 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)

Interrogez la table partitionnée

1. Ouvrez la console Athena.
2. Exécutez la commande suivante pour interroger la table et confirmer qu'elle est partitionnée :

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