Test di uno schema

Durante lo sviluppo del codice, è necessario eseguire test locali per verificare che il layout del flusso di lavoro sia corretto.

Il test locale non genera processi, crawler o attivazioni AWS Glue. Invece, esegui lo script di layout localmente e utilizzi i metodi to_json() e validate() per stampare gli oggetti e trovare gli errori. Questi metodi sono disponibili in tutte e tre le classi definite nelle librerie.

Sono disponibili due modi per gestire gli argomenti user_params e system_params cheAWS Glue passa alla funzione di layout. Il codice test-bench può creare un dizionario di valori di esempio dei parametri del piano e passarli alla funzione di layout come argomento user_params. In alternativa, puoi rimuovere i riferimenti a user_params e sostituirli con stringhe hardcoded.

Se il tuo codice utilizza proprietà region e accountId nell'argomento system_params, puoi passare nel tuo dizionario per system_params.

Per testare un piano

1. Avvia un interprete Python in una directory con le librerie o carica i file del piano e le librerie fornite nel tuo ambiente di sviluppo integrato (IDE) preferito.

2. Assicurati che il tuo codice importi le librerie fornite.

3. Aggiungi codice alla tua funzione di layout per chiamare validate() o to_json() su qualsiasi entità o sull'oggetto Workflow. Ad esempio, se il codice crea un oggetto Crawler denominato mycrawler, è possibile chiamare validate() come segue.

   mycrawler.validate()

   Puoi stampare mycrawler come segue:

   print(mycrawler.to_json())

   Se chiami to_json su un oggetto, non è necessario chiamare anche validate(), perché to_json() chiama validate().

   È molto utile chiamare questi metodi sull'oggetto flusso di lavoro. Supponendo che lo script denomini l'oggetto flusso di lavoro my_workflow, convalida e stampa l'oggetto flusso di lavoro come segue.

   print(my_workflow.to_json())

   Per ulteriori informazioni su to_json() e validate(), consulta Metodi di classe.

   Puoi anche importare pprint e stampare con precisione l'oggetto flusso di lavoro, come illustrato nell'esempio più avanti in questa sezione.

4. Esegui il codice, correggi gli errori e infine rimuovi tutte le chiamate a validate() o to_json().

L'esempio seguente mostra come costruire un dizionario di parametri di esempio del piano e passarli come argomento user_params alla funzione di layout generate_compaction_workflow. Viene inoltre illustrato come stampare con precisione l'oggetto flusso di lavoro generato.

from pprint import pprint
from awsglue.blueprint.workflow import *
from awsglue.blueprint.job import *
from awsglue.blueprint.crawler import *
 
USER_PARAMS = {"WorkflowName": "compaction_workflow",
               "ScriptLocation": "s3://awsexamplebucket1/scripts/threaded-compaction.py",
               "PassRole": "arn:aws:iam::111122223333:role/GlueRole-ETL",
               "DatabaseName": "cloudtrial",
               "TableName": "ct_cloudtrail",
               "CoalesceFactor": 4,
               "MaxThreadWorkers": 200}
 
 
def generate_compaction_workflow(user_params: dict, system_params: dict) -> Workflow:
    compaction_job = Job(Name=f"{user_params['WorkflowName']}_etl_job",
                         Command={"Name": "glueetl",
                                  "ScriptLocation": user_params['ScriptLocation'],
                                  "PythonVersion": "3"},
                         Role="arn:aws:iam::111122223333:role/AWSGlueServiceRoleDefault",
                         DefaultArguments={"DatabaseName": user_params['DatabaseName'],
                                           "TableName": user_params['TableName'],
                                           "CoalesceFactor": user_params['CoalesceFactor'],
                                           "max_thread_workers": user_params['MaxThreadWorkers']})
 
    catalog_target = {"CatalogTargets": [{"DatabaseName": user_params['DatabaseName'], "Tables": [user_params['TableName']]}]}
 
    compacted_files_crawler = Crawler(Name=f"{user_params['WorkflowName']}_post_crawl",
                                      Targets = catalog_target,
                                      Role=user_params['PassRole'],
                                      DependsOn={compaction_job: "SUCCEEDED"},
                                      WaitForDependencies="AND",
                                      SchemaChangePolicy={"DeleteBehavior": "LOG"})
 
    compaction_workflow = Workflow(Name=user_params['WorkflowName'],
                                   Entities=Entities(Jobs=[compaction_job],
                                                     Crawlers=[compacted_files_crawler]))
    return compaction_workflow
 
generated = generate_compaction_workflow(user_params=USER_PARAMS, system_params={})
gen_dict = generated.to_json()
 
pprint(gen_dict)