Menguji cetak biru

Sementara Anda mengembangkan kode Anda, Anda harus melakukan pengujian secara lokal untuk memverifikasi bahwa tata letak alur kerja sudah benar.

Pengujian lokal tidak menghasilkan AWS Glue pekerjaan, crawler, atau pemicu. Sebaliknya, Anda jalankan skrip tata letak secara lokal dan gunakan metode to_json() dan validate() untuk mencetak objek dan menemukan kesalahan. Metode ini tersedia di semua tiga kelas yang ditentukan dalam perpustakaan.

Ada dua cara untuk menangani user_params dan system_params argumen yang AWS Glue lolos ke fungsi tata letak Anda. Kode test-bench Anda dapat membuat sebuah kamus sampel nilai parameter cetak biru dan memberikannya untuk fungsi tata letak sebagai argumen user_params. Atau, Anda dapat menghapus referensi ke user_params dan menggantinya dengan string yang di-hardcoding.

Jika kode Anda menggunakan properti region dan accountId di system_params, Anda dapat memberikan dalam kamus Anda sendiri untuk system_params.

Untuk menguji sebuah cetak biru

1. Mulai sebuah interpreter Python dalam direktori yang memiliki perpustakaan, atau muat file cetak biru dan perpustakaan yang disediakan ke lingkungan pengembangan terpadu (IDE) pilihan Anda.

2. Pastikan bahwa kode Anda mengimpor perpustakaan yang disediakan.

3. Tambahkan kode ke fungsi tata letak Anda untuk memanggil validate() atau to_json() pada entitas apa pun atau objek Workflow. Sebagai contoh, jika kode Anda membuat objek Crawler bernama mycrawler, maka Anda dapat memanggil validate() sebagai berikut.

   mycrawler.validate()

   Anda dapat mencetak mycrawler sebagai berikut:

   print(mycrawler.to_json())

   Jika Anda memanggil to_json pada sebuah objek, maka Anda tidak perlu juga memanggil validate(), karena to_json() memanggil validate().

   Hal ini akan sangat berguna untuk memanggil metode ini pada objek alur kerja. Dengan asumsi bahwa skrip Anda memberi nama objek alur kerja my_workflow, validasi dan cetak objek alur kerja sebagai berikut.

   print(my_workflow.to_json())

   Untuk informasi selengkapnya tentang to_json() dan validate(), lihat Metode kelas.

   Anda juga dapat mengimpor pprint dan melakukan pretty-print pada objek alur kerja, seperti yang ditunjukkan pada contoh nanti di bagian ini.

4. Jalankan kode, perbaiki kesalahan, dan akhirnya hapus setiap panggilan ke validate() atau to_json().

Contoh berikut menunjukkan cara membangun kamus parameter cetak biru sampel dan menyebarkannya sebagai argumen user_params ke fungsi tata letak generate_compaction_workflow. Hal ini juga menunjukkan cara melakukan pretty-print pada objek alur kerja yang dihasilkan.

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)