블루프린트 테스트 - AWS Glue

블루프린트 테스트

코드를 개발하는 동안 워크플로 레이아웃이 올바른지 확인하기 위해 로컬 테스트를 수행해야 합니다.

로컬 테스트는 AWS Glue 작업, 크롤러 또는 트리거를 생성하지 않습니다. 대신 레이아웃 스크립트를 로컬에서 실행하고 to_json() 및 validate() 메서드를 사용하여 객체를 인쇄하고 오류를 찾습니다. 이러한 방법은 라이브러리에 정의 된 3가지 클래스 모두에서 사용할 수 있습니다.

AWS Glue가 레이아웃 함수에 전달하는 user_params 및 system_params 인수를 처리하는 두 가지 방법이 있습니다. 테스트 벤치 코드는 샘플 블루프린트 파라미터 값의 딕셔너리를 생성하고 이를 user_params 인수로 레이아웃 함수에 전달할 수 있습니다. 또는 user_params에 대한 참조를 제거하고 하드코딩된 문자열로 바꿀 수 있습니다.

코드에서 system_params 인수의 region 및 accountId 속성을 사용하는 경우 system_params에 대한 딕셔너리를 전달할 수 있습니다.

블루프린트를 테스트하려면

  1. 라이브러리가 있는 디렉터리에서 Python 인터프리터를 시작하거나 블루프린트 파일과 제공된 라이브러리를 원하는 IDE(통합 개발 환경)로 로드합니다.

  2. 코드가 제공된 라이브러리를 가져 오는지 확인합니다.

  3. 레이아웃 기능에 코드를 추가하여 엔터티 또는 Workflow 객체에서 validate() 또는 to_json()을 호출합니다. 예를 들어 코드가 mycrawler라는 Crawler 객체를 생성하는 경우 다음과 같이 validate()를 호출할 수 있습니다.

    mycrawler.validate()

    다음과 같이 mycrawler를 인쇄 할 수 있습니다.

    print(mycrawler.to_json())

    객체에서 to_json을 호출하면 to_json()이 validate()를 호출하기 때문에 validate()도 호출할 필요가 없습니다.

    워크플로 객체에서 이러한 메서드를 호출하는 것이 가장 유용합니다. 스크립트가 워크플로 객체의 이름을 my_workflow라고 가정하고 다음과 같이 워크플로 객체를 검증하고 인쇄합니다.

    print(my_workflow.to_json())

    to_json() 및 validate()에 대한 자세한 내용은 클래스 메서드섹션을 참조하세요.

    이 섹션의 뒷부분에 나오는 예제와 같이 pprint를 가져와서 워크플로 객체를 예쁘게 인쇄할 수도 있습니다.

  4. 코드를 실행하고, 오류를 수정하고, 마지막으로 validate() 또는 to_json()에 대한 호출을 제거합니다.

다음 예제는 샘플 블루프린트 파라미터의 딕셔너리를 구성하고 레이아웃 함수 generate_compaction_workflow에 user_params 인수로 전달하는 방법을 보여줍니다. 또한 생성된 워크플로 객체를 예쁘게 인쇄하는 방법을 보여줍니다.

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)