Mise à jour du schéma et ajout de nouvelles partitions dans le catalogue de données à l'aide de AWS Glue tâches ETL - AWS Glue
Nouvelles partitionsMise à jour du schéma de tableCréation de tablesRestrictions

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Mise à jour du schéma et ajout de nouvelles partitions dans le catalogue de données à l'aide de AWS Glue tâches ETL

Votre travail d'extraction, de transformation et de chargement (ETL) peut créer de nouvelles partitions de table dans le magasin de données cible. Le schéma de votre jeu de données peut évoluer et diverger par rapport au AWS Glue Schéma du catalogue de données au fil du temps. AWS Glue Les tâches ETL fournissent désormais plusieurs fonctions que vous pouvez utiliser dans votre script ETL pour mettre à jour votre schéma et vos partitions dans le catalogue de données. Ces fonctions vous permettent de voir les résultats de votre tâche ETL dans le catalogue de données, sans avoir à exécuter à nouveau l'crawler.

Nouvelles partitions

Si vous souhaitez afficher les nouvelles partitions dans le AWS Glue Data Catalog, vous pouvez effectuer l'une des opérations suivantes :

  • Une fois la tâche terminée, réexécutez l'crawler pour afficher les nouvelles partitions sur la console.

  • Une fois la tâche terminée, affichez immédiatement les nouvelles partitions sur la console, sans avoir à réexécuter l'crawler. Pour activer cette fonctionnalité, ajoutez quelques lignes de code au script ETL, comme illustré dans les exemples suivants. Le code utilise l'argument enableUpdateCatalog pour indiquer que le catalogue de données doit être mis à jour pendant l'exécution de la tâche lors de la création des partitions.

Méthode 1 :

Transmettez enableUpdateCatalog et partitionKeys dans un argument d'options.

Python
additionalOptions = {"enableUpdateCatalog": True}
additionalOptions["partitionKeys"] = ["region", "year", "month", "day"]


sink = glueContext.write_dynamic_frame_from_catalog(frame=last_transform, database=<target_db_name>,
                                                    table_name=<target_table_name>, transformation_ctx="write_sink",
                                                    additional_options=additionalOptions)
Scala
val options = JsonOptions(Map(
    "path" -> <S3_output_path>, 
    "partitionKeys" -> Seq("region", "year", "month", "day"), 
    "enableUpdateCatalog" -> true))
val sink = glueContext.getCatalogSink(
    database = <target_db_name>, 
    tableName = <target_table_name>, 
    additionalOptions = options)sink.writeDynamicFrame(df)
Méthode 2 :

Transmettez enableUpdateCatalog et partitionKeys dans getSink() et appelez setCatalogInfo() au niveau de l'objet DataSink.

Python
sink = glueContext.getSink(
    connection_type="s3", 
    path="<S3_output_path>",
    enableUpdateCatalog=True,
    partitionKeys=["region", "year", "month", "day"])
sink.setFormat("json")
sink.setCatalogInfo(catalogDatabase=<target_db_name>, catalogTableName=<target_table_name>)
sink.writeFrame(last_transform)
Scala
val options = JsonOptions(
   Map("path" -> <S3_output_path>, 
       "partitionKeys" -> Seq("region", "year", "month", "day"), 
       "enableUpdateCatalog" -> true))
val sink = glueContext.getSink("s3", options).withFormat("json")
sink.setCatalogInfo(<target_db_name>, <target_table_name>)
sink.writeDynamicFrame(df)

Vous pouvez désormais créer de nouvelles tables de catalogue, mettre à jour des tables existantes avec un schéma modifié et ajouter de nouvelles partitions de table dans le catalogue de données à l'aide d'un AWS Glue Job ETL lui-même, sans qu'il soit nécessaire de réexécuter les robots d'exploration.

Mise à jour du schéma de table

Si vous souhaitez remplacer le schéma de la table du catalogue de données, vous pouvez effectuer l'une des opérations suivantes :

  • Une fois la tâche terminée, exécutez à nouveau l’crawler et assurez-vous que celui-ci est configuré pour mettre à jour également la définition de la table. Affichez les nouvelles partitions sur la console ainsi que les mises à jour de schéma lorsque l'crawler se termine. Pour plus d'informations, consultez Configuration d'un crawler utilisant l'API.

  • Une fois la tâche terminée, affichez immédiatement le schéma modifié sur la console, sans avoir à exécuter de nouveau l'crawler. Pour activer cette fonctionnalité, ajoutez quelques lignes de code au script ETL, comme illustré dans les exemples suivants. Le code utilise enableUpdateCatalog défini sur true et updateBehavior défini sur UPDATE_IN_DATABASE, ce qui indique que le schéma doit être remplacé et que de nouvelles partitions doivent être ajoutées au catalogue de données pendant l'exécution de la tâche.

Python
additionalOptions = {
    "enableUpdateCatalog": True, 
    "updateBehavior": "UPDATE_IN_DATABASE"}
additionalOptions["partitionKeys"] = ["partition_key0", "partition_key1"]

sink = glueContext.write_dynamic_frame_from_catalog(frame=last_transform, database=<dst_db_name>,
    table_name=<dst_tbl_name>, transformation_ctx="write_sink",
    additional_options=additionalOptions)
job.commit()
Scala
val options = JsonOptions(Map(
    "path" -> outputPath, 
    "partitionKeys" -> Seq("partition_0", "partition_1"), 
    "enableUpdateCatalog" -> true))
val sink = glueContext.getCatalogSink(database = nameSpace, tableName = tableName, additionalOptions = options)
sink.writeDynamicFrame(df)

Vous pouvez également définir la valeur de updateBehavior sur LOG si vous souhaitez empêcher le remplacement du schéma de table, mais que vous souhaitez toujours ajouter les nouvelles partitions. Comme la valeur par défaut de updateBehavior est UPDATE_IN_DATABASE, si vous ne le définissez pas explicitement, le schéma de table est remplacé.

Si enableUpdateCatalog n'a pas la valeur true, quelle que soit l'option sélectionnée pour updateBehavior, la tâche ETL ne mettra pas à jour la table dans le catalogue de données.

Création de tables

Vous pouvez également utiliser les mêmes options pour créer une table dans le catalogue de données. Vous pouvez spécifier la base de données et le nouveau nom de table avec setCatalogInfo.

Python
sink = glueContext.getSink(connection_type="s3", path="s3://path/to/data",
    enableUpdateCatalog=True, updateBehavior="UPDATE_IN_DATABASE",
    partitionKeys=["partition_key0", "partition_key1"])
sink.setFormat("<format>")
sink.setCatalogInfo(catalogDatabase=<dst_db_name>, catalogTableName=<dst_tbl_name>)
sink.writeFrame(last_transform)
Scala
val options = JsonOptions(Map(
    "path" -> outputPath, 
    "partitionKeys" -> Seq("<partition_1>", "<partition_2>"), 
    "enableUpdateCatalog" -> true, 
    "updateBehavior" -> "UPDATE_IN_DATABASE"))
val sink = glueContext.getSink(connectionType = "s3", connectionOptions = options).withFormat("<format>")
sink.setCatalogInfo(catalogDatabase = “<dst_db_name>”, catalogTableName = “<dst_tbl_name>”)
sink.writeDynamicFrame(df)

Restrictions

Prenez note des restrictions suivantes :

  • Seules des cibles Amazon Simple Storage Service (Amazon S3) sont prises en charge.

  • La fonctionnalité enableUpdateCatalog n'est pas prise en charge pour les tables gouvernées.

  • Seuls les formats suivants sont pris en charge : json, csv, avro et parquet.

  • Pour créer ou mettre à jour des tables avec la parquet classification, vous devez utiliser AWS Glue graveur de parquet optimisé pour DynamicFrames. Cela peut être réalisé à l'aide d'un des moyens suivants :

    • Si vous mettez à jour une table existante dans le catalogue avec la classification parquet, la propriété de la table "useGlueParquetWriter" doit être définie sur true avant que vous ne la mettiez à jour. Vous pouvez définir cette propriété via le AWS Glue APIs /SDK, via la console ou via une instruction Athena DDL.

      Champ d'édition des propriétés de la table de catalogue dans AWS Glue la console.

      Une fois la propriété de la table de catalogue définie, vous pouvez utiliser l'extrait de code suivant pour mettre à jour la table de catalogue avec les nouvelles données :

      glueContext.write_dynamic_frame.from_catalog(
    frame=frameToWrite,
    database="dbName",
    table_name="tableName",
    additional_options={
        "enableUpdateCatalog": True,
        "updateBehavior": "UPDATE_IN_DATABASE"
    }
)

    • Si la table n'existe pas déjà dans le catalogue, vous pouvez utiliser la méthode getSink() de votre script avec connection_type="s3" pour ajouter la table et ses partitions au catalogue, ainsi que pour écrire les données sur Amazon S3. Fournissez les partitionKeys et la compression appropriés pour votre flux de travail.

      s3sink = glueContext.getSink(
    path="s3://bucket/folder/",
    connection_type="s3",
    updateBehavior="UPDATE_IN_DATABASE",
    partitionKeys=[],
    compression="snappy",
    enableUpdateCatalog=True
)
    
s3sink.setCatalogInfo(
    catalogDatabase="dbName", catalogTableName="tableName"
)
    
s3sink.setFormat("parquet", useGlueParquetWriter=True)
s3sink.writeFrame(frameToWrite)

    • La valeur de glueparquet format est une ancienne méthode permettant d'activer le rédacteur de AWS Glue parquet.

  • Quand updateBehavior a la valeur LOG, de nouvelles partitions sont ajoutées uniquement si le schéma DynamicFrame est équivalent ou qu'il contient un sous-ensemble des colonnes définies dans le schéma de la table du catalogue de données.

  • Les mises à jour du schéma ne sont pas prises en charge pour les tables non partitionnées (qui n'utilisent pas l'option « partitionKeys »).

  • Vos clés de partition (partitionKeys) doivent être équivalentes, et dans le même ordre, entre votre paramètre passé dans votre script ETL et les clés de partition (partitionKeys) du schéma de votre table du catalogue de données.

  • Cette fonction ne prend pas encore en charge la mise à jour/création de tables dans lesquelles les schémas de mise à jour sont imbriqués (par exemple, les tableaux à l'intérieur de structures).

Pour de plus amples informations, veuillez consulter Programmation de scripts Spark.