Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà. # Aggiornamento dello schema e aggiunta di nuove partizioni nel catalogo dati usando processi ETL AWS Glue Il processo di estrazione, trasformazione e caricamento (ETL) potrebbe creare nuove partizioni di tabella nell'archivio dati di destinazione. Lo schema del set di dati può evolversi e divergere dallo schema del catalogo dati AWS Glue nel corso del tempo. AWS Glue I processi ETL ora forniscono diverse funzionalità che puoi utilizzare all'interno dello script ETL per aggiornare lo schema e le partizioni nel catalogo dati. Queste caratteristiche ti consentono di vedere i risultati del processo ETL nel catalogo dati, senza dover eseguire nuovamente il crawler. ## Nuove partizioni Se desideri visualizzare le nuove partizioni in AWS Glue Data Catalog, puoi effettuare una delle seguenti operazioni: + Al termine del processo, esegui nuovamente il crawler e visualizza le nuove partizioni sulla console al termine del crawler. + Al termine del processo, visualizza immediatamente le nuove partizioni sulla console, senza dover eseguire nuovamente il crawler. Puoi abilitare questa caratteristica aggiungendo alcune righe di codice allo script ETL, come mostrato negli esempi seguenti. Il codice utilizza l'argomento `enableUpdateCatalog` per indicare che il catalogo dati deve essere aggiornato durante l'esecuzione del processo quando vengono create nuove partizioni. **Metodo 1** Passare `enableUpdateCatalog` e `partitionKeys` in un argomento opzioni. ``` additionalOptions = {"enableUpdateCatalog": True} additionalOptions["partitionKeys"] = ["region", "year", "month", "day"] sink = glueContext.write_dynamic_frame_from_catalog(frame=last_transform, database=, table_name=, transformation_ctx="write_sink", additional_options=additionalOptions) ``` ``` val options = JsonOptions(Map( "path" -> , "partitionKeys" -> Seq("region", "year", "month", "day"), "enableUpdateCatalog" -> true)) val sink = glueContext.getCatalogSink( database = , tableName = , additionalOptions = options)sink.writeDynamicFrame(df) ``` **Metodo 2** Passare `enableUpdateCatalog` e `partitionKeys` in `getSink()` e chiamare `setCatalogInfo()` sull'oggetto `DataSink`. ``` sink = glueContext.getSink( connection_type="s3", path="", enableUpdateCatalog=True, partitionKeys=["region", "year", "month", "day"]) sink.setFormat("json") sink.setCatalogInfo(catalogDatabase=, catalogTableName=) sink.writeFrame(last_transform) ``` ``` val options = JsonOptions( Map("path" -> , "partitionKeys" -> Seq("region", "year", "month", "day"), "enableUpdateCatalog" -> true)) val sink = glueContext.getSink("s3", options).withFormat("json") sink.setCatalogInfo(, ) sink.writeDynamicFrame(df) ``` Ora, puoi creare nuove tabelle di catalogo, aggiornare le tabelle esistenti con schema modificato e aggiungere nuove partizioni di tabella nel catalogo dati utilizzando direttamente un processo ETL di AWS Glue, senza la necessità di eseguire nuovamente i crawler. ## Aggiornamento dello schema della tabella Se desideri sovrascrivere lo schema della tabella del catalogo dati, puoi eseguire una delle seguenti operazioni: + Al termine del processo, esegui nuovamente il crawler e assicurati che il crawler sia configurato per aggiornare anche la definizione della tabella. Visualizza le nuove partizioni sulla console insieme agli eventuali aggiornamenti dello schema, al termine del crawler. Per maggiori informazioni, consulta [Configurazione di un crawler utilizzando l'API](https://docs.aws.amazon.com/glue/latest/dg/crawler-configuration.html#crawler-configure-changes-api). + Al termine del processo, visualizza immediatamente lo schema modificato sulla console, senza dover eseguire nuovamente il crawler. Puoi abilitare questa caratteristica aggiungendo alcune righe di codice allo script ETL, come mostrato negli esempi seguenti. Il codice utilizza `enableUpdateCatalog` impostato su true, e anche `updateBehavior` impostato su `UPDATE_IN_DATABASE`, il che indica di sovrascrivere lo schema e aggiungere nuove partizioni nel catalogo dati durante l'esecuzione del processo. ------ #### [ 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=, table_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) ``` ------ Puoi inoltre impostare il valore `updateBehavior` su `LOG` se desideri impedire che lo schema di tabella venga sovrascritto, ma se desidera comunque aggiungere le nuove partizioni. Il valore predefinito di `updateBehavior` è `UPDATE_IN_DATABASE`, quindi se non lo definisci esplicitamente, lo schema della tabella verrà sovrascritto. Se `enableUpdateCatalog` non è impostato su true, indipendentemente da qualsiasi opzione selezionata per `updateBehavior`, il processo ETL non aggiornerà la tabella nel catalogo dati. ## Creazione di nuove tabelle Puoi inoltre utilizzare le stesse opzioni per creare una nuova tabella nel catalogo dati. Puoi specificare il database e il nome della nuova tabella utilizzando `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("") sink.setCatalogInfo(catalogDatabase=, catalogTableName=) sink.writeFrame(last_transform) ``` ------ #### [ Scala ] ``` val options = JsonOptions(Map( "path" -> outputPath, "partitionKeys" -> Seq("", ""), "enableUpdateCatalog" -> true, "updateBehavior" -> "UPDATE_IN_DATABASE")) val sink = glueContext.getSink(connectionType = "s3", connectionOptions = options).withFormat("") sink.setCatalogInfo(catalogDatabase = “”, catalogTableName = “”) sink.writeDynamicFrame(df) ``` ------ ## Restrizioni Prestare attenzione alle seguenti restrizioni: + Sono supportate solo le destinazioni di Amazon Simple Storage Service (Amazon S3). + La funzionalità `enableUpdateCatalog` non è supportata per le tabelle governate. + Sono supportati solo i seguenti formati: `json`, `csv`, `avro`, e `parquet`. + Per creare o aggiornare tabelle con la `parquet` classificazione, è necessario utilizzare il parquet writer AWS Glue ottimizzato per. DynamicFrames È possibile farlo in uno dei modi seguenti: + Se stai aggiornando una tabella esistente nel catalogo con la classificazione `parquet`, la proprietà della tabella `"useGlueParquetWriter"` deve essere impostata su `true` prima di aggiornarla. È possibile impostare questa proprietà tramite AWS Glue APIs /SDK, tramite la console o tramite un'istruzione Athena DDL. ![\[Campo di modifica delle proprietà della tabella del catalogo nella console. AWS Glue\]](http://docs.aws.amazon.com/it_it/glue/latest/dg/images/edit-table-property.png) Una volta impostata la proprietà della tabella del catalogo, puoi utilizzare il seguente frammento di codice per aggiornare la tabella del catalogo con i nuovi dati: ``` glueContext.write_dynamic_frame.from_catalog( frame=frameToWrite, database="dbName", table_name="tableName", additional_options={ "enableUpdateCatalog": True, "updateBehavior": "UPDATE_IN_DATABASE" } ) ``` + Se la tabella non esiste ancora nel catalogo, puoi utilizzare il metodo `getSink()` nello script con `connection_type="s3"` per aggiungere la tabella e le sue partizioni al catalogo, oltre a scrivere i dati su Amazon S3. Fornisci i valori appropriati di `partitionKeys` e `compression` per il tuo flusso di lavoro. ``` 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) ``` + Il valore di `glueparquet` formato è un metodo obsoleto per abilitare il AWS Glue parquet writer. + Quando `updateBehavior` è impostato su `LOG`, nuove partizioni verranno aggiunte solo se lo schema `DynamicFrame` è equivalente o contiene un sottoinsieme delle colonne definite nello schema della tabella del catalogo dati. + Gli aggiornamenti dello schema non sono supportati per le tabelle non partizionate (che non utilizzano l'opzione “partitionKeys”). + Le PartitionKeys devono essere equivalenti, e nello stesso ordine, tra il parametro passato nello script ETL e le PartitionKeys nello schema della tabella del catalogo dati. + Questa funzionalità attualmente non supporta ancora le updating/creating tabelle in cui sono annidati gli schemi di aggiornamento (ad esempio, gli array all'interno delle strutture). Per ulteriori informazioni, consulta [Script di programmazione Spark](aws-glue-programming.md). # Lavorare con le connessioni MongoDB nei processi ETL Puoi creare una connessione per MongoDB e usare quella connessione nel tuo processo AWS Glue. Per ulteriori informazioni, consultare [Connessioni MongoDB](aws-glue-programming-etl-connect-mongodb-home.md) la guida alla AWS Glue programmazione. La connessione`url`, `username` e `password` sono archiviati nella connessione MongoDB. Altre opzioni possono essere specificate nello script del processo ETL utilizzando il parametro `additionalOptions` di `glueContext.getCatalogSource`. Le altre opzioni possono includere: + `database`: (Obbligatorio) Il database MongoDB da cui leggere. + `collection`: (Obbligatorio) La raccolta MongoDB da cui leggere. Posizionando le informazioni di `database` e `collection` all'interno dello script del processo ETL, puoi utilizzare la stessa connessione in più processi. 1. Crea una AWS Glue Data Catalog connessione per l'origine dati MongoDB. Consulta ["connectionType": "mongodb"](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-connect.html#aws-glue-programming-etl-connect-mongodb) per una descrizione dei parametri di connessione. È possibile creare la connessione utilizzando la console APIs o la CLI. 1. Crea un database in AWS Glue Data Catalog per memorizzare le definizioni delle tabelle per i tuoi dati MongoDB. Per ulteriori informazioni, consulta [Creazione di database](define-database.md). 1. Crea un crawler che esegue il crawling dati in MongoDB utilizzando le informazioni nella connessione per connettersi a MongoDB. Il crawler crea le tabelle AWS Glue Data Catalog che descrivono le tabelle del database MongoDB che usi nel tuo job. Per ulteriori informazioni, consulta [Utilizzo dei crawler per compilare il Catalogo dati](add-crawler.md). 1. Crea un processo con uno script personalizzato. È possibile creare il lavoro utilizzando la console APIs o la CLI. Per ulteriori informazioni, consulta [Aggiunta di processi in AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/add-job.html). 1. Scegli le destinazioni dati per il tuo processo. Le tabelle che rappresentano la destinazione dei dati possono essere definite nel catalogo dati oppure il processo può creare le tabelle di destinazione quando viene eseguito. Puoi scegliere una posizione di destinazione al momento della creazione del processo. Se la destinazione richiede una connessione, anche la connessione ha un riferimento nel tuo processo. Se il processo richiede più destinazioni dati, in seguito potrai aggiungerle modificando lo script. 1. Personalizza l'ambiente di elaborazione del processo fornendo gli argomenti per il tuo processo e lo script generato. Di seguito è illustrato un esempio di creazione di un `DynamicFrame` dal database MongoDB in base alla struttura della tabella definita nel catalogo dati. Il codice utilizza `additionalOptions` per fornire le informazioni aggiuntive sull'origine dati: ------ #### [ Scala ] ``` val resultFrame: DynamicFrame = glueContext.getCatalogSource( database = catalogDB, tableName = catalogTable, additionalOptions = JsonOptions(Map("database" -> DATABASE_NAME, "collection" -> COLLECTION_NAME)) ).getDynamicFrame() ``` ------ #### [ Python ] ``` glue_context.create_dynamic_frame_from_catalog( database = catalogDB, table_name = catalogTable, additional_options = {"database":"database_name", "collection":"collection_name"}) ``` ------ 1. Esegui il processo, on demand o tramite un trigger.