Airflow 작업 로그를 사용하여 매니페스트 수집 문제 해결

아티클
02/23/2024

이 문서는 Airflow 작업 로그를 사용하여 Azure Data Manager for Energy에서 매니페스트 수집과 관련된 워크플로 문제를 해결하는 데 도움이 됩니다.

매니페스트 수집 DAG 워크플로 유형

매니페스트 수집을 위한 DAG(Directed Acyclic Graph) 워크플로에는 단일 매니페스트 및 일괄 처리 업로드의 두 가지 유형이 있습니다.

단일 매니페스트

단일 매니페스트 파일은 매니페스트 수집 워크플로를 트리거하는 데 사용됩니다.

DagTaskName 값	설명
`update_status_running_task`	워크플로 서비스를 호출하고 DAG의 상태를 데이터베이스에서 `running`으로 표시합니다.
`check_payload_type`	수집 유형이 일괄 처리인지 아니면 단일 매니페스트인지 검사합니다.
`validate_manifest_schema_task`	매니페스트에 언급된 모든 스키마 형식이 있고 참조 스키마 무결성이 있는지 확인합니다. 모든 잘못된 값이 매니페스트에서 제거됩니다.
`provide_manifest_integrity_task`	OSDU® R3 매니페스트 내에서 참조의 유효성을 검사하고 잘못된 엔터티를 제거합니다. 이 연산자는 부모/자식 유효성 검사를 담당합니다. 모든 고아와 유사한 엔터티는 로깅되고 유효성이 검사된 매니페스트에서 제외됩니다. 외부 참조 레코드가 검색됩니다. 찾을 수 없는 경우 매니페스트 엔터티가 삭제됩니다. 모든 서로게이트 키 참조도 확인됩니다.
`process_single_manifest_file_task`	이전 단계에서 가져온 최종 매니페스트 엔터티의 수집을 수행합니다. 데이터 레코드는 스토리지 서비스를 통해 수집됩니다.
`update_status_finished_task`	워크플로 서비스를 호출하고 DAG의 상태를 데이터베이스에서 `finished` 또는 `failed`로 표시합니다.

일괄 업로드

여러 매니페스트 파일은 동일한 워크플로 서비스 요청의 일부입니다. 요청 페이로드의 매니페스트 섹션은 항목 사전 대신 목록입니다.

DagTaskName 값	설명
`update_status_running_task`	워크플로 서비스를 호출하고 DAG의 상태를 데이터베이스에서 `running`으로 표시합니다.
`check_payload_type`	수집 유형이 일괄 처리인지 아니면 단일 매니페스트인지 검사합니다.
`batch_upload`	매니페스트 목록을 병렬로 처리할 3개의 일괄 처리로 나눕니다. (내보낸 작업 로그가 없습니다.)
`process_manifest_task_(1 / 2 / 3)`	매니페스트 목록을 3개의 그룹으로 나누고 처리합니다. `validate_manifest_schema_task`, `provide_manifest_integrity_task`, `process_single_manifest_file_task`에서 수행되는 모든 단계는 이러한 작업에서 순차적으로 압축되고 수행됩니다.
`update_status_finished_task`	워크플로 서비스를 호출하고 DAG의 상태를 데이터베이스에서 `finished` 또는 `failed`로 표시합니다.

페이로드 유형(단일 또는 일괄 처리)에 따라 check_payload_type 작업은 적절한 분기를 선택하고 다른 분기의 작업을 건너뜁니다.

필수 조건

Airflow 작업 로그를 Azure Monitor와 통합해야 합니다. Azure Monitor와 Airflow 로그 통합을 참조하세요.

문제를 디버그할 수 있도록 다음 열이 Airflow 작업 로그에 표시됩니다.

매개 변수 이름	설명
`RunID`	트리거된 DAG 실행의 고유 실행 ID입니다.
`CorrelationID`	DAG 실행의 고유 상관 관계 ID(실행 ID와 동일).
`DagName`	DAG 워크플로 이름. 예를 들어 `Osdu_ingest`는 매니페스트 수집의 워크플로 이름입니다.
`DagTaskName`	DAG 워크플로의 작업 이름입니다. 예를 들어 `update_status_running_task`는 매니페스트 수집의 작업 이름입니다.
`Content`	작업 실행 중에 Airflow에서 내보내는 오류 로그 메시지(오류 또는 예외)입니다.
`LogTimeStamp`	DAG 실행의 시간 간격입니다.
`LogLevel`	오류의 수준입니다. 값은 `DEBUG`, `INFO`, `WARNING`, `ERROR`입니다. `ERROR` 수준에서 필터링하여 대부분의 예외 및 오류 메시지를 볼 수 있습니다.

실패한 DAG 실행

Update_status_running_task 또는 Update_status_finished_task에서 워크플로 실행이 실패했으며 데이터 레코드가 수집되지 않았습니다.

가능한 이유

데이터 파티션 ID가 잘못되어 파티션 API 호출이 인증되지 않았습니다.
요청 본문의 실행 컨텍스트에서 키 이름이 잘못되었습니다.
워크플로 서비스가 실행되고 있지 않거나 5xx 오류가 발생합니다.

워크플로 상태

워크플로 상태가 failed로 표시됩니다.

솔루션

update_status_running_task 또는 update_status_finished_task에 대한 Airflow 작업 로그를 확인합니다. 올바른 데이터 파티션 ID 또는 키 이름을 전달하여 페이로드를 수정합니다.

샘플 Kusto 쿼리:

    OEPAirFlowTask
        | where DagName == "Osdu_ingest"
        | where DagTaskName == "update_status_running_task"
        | where LogLevel == "ERROR" // ERROR/DEBUG/INFO/WARNING
        | where RunID == '<run_id>'

샘플 추적 출력:

    [2023-02-05, 12:21:54 IST] {taskinstance.py:1703} ERROR - Task failed with exception
    Traceback (most recent call last):
      File "/home/airflow/.local/lib/python3.8/site-packages/osdu_ingestion/libs/context.py", line 50, in populate
        data_partition_id = ctx_payload['data-partition-id']
    KeyError: 'data-partition-id'
    
    requests.exceptions.HTTPError: 403 Client Error: Forbidden for url: https://contoso.energy.azure.com/api/workflow/v1/workflow/Osdu_ingest/workflowRun/e9a815f2-84f5-4513-9825-4d37ab291264

실패한 스키마 유효성 검사

스키마 유효성 검사에 실패하여 레코드가 수집되지 않았습니다.

가능한 이유

스키마 서비스에서 "스키마를 찾을 수 없음" 오류가 발생합니다.
매니페스트 본문이 스키마 형식을 준수하지 않습니다.
스키마 참조가 잘못되었습니다.
스키마 서비스에서 5xx 오류를 throw합니다.

워크플로 상태

워크플로 상태가 finished로 표시됩니다. 잘못된 엔터티를 건너뛰고 수집을 계속하기 때문에 워크플로 상태 오류가 표시되지 않습니다.

솔루션

validate_manifest_schema_task 또는 process_manifest_task에 대한 Airflow 작업 로그를 확인합니다. 올바른 데이터 파티션 ID 또는 키 이름을 전달하여 페이로드를 수정합니다.

샘플 Kusto 쿼리:

    OEPAirFlowTask
    | where DagName has "Osdu_ingest"
    | where DagTaskName == "validate_manifest_schema_task" or DagTaskName has "process_manifest_task"
    | where LogLevel == "ERROR"
    | where RunID == "<run_id>"
    | order by ['time'] asc

샘플 추적 출력:

    Error traces to look out for
    [2023-02-05, 14:55:37 IST] {connectionpool.py:452} DEBUG - https://contoso.energy.azure.com:443 "GET /api/schema-service/v1/schema/osdu:wks:work-product-component--WellLog:2.2.0 HTTP/1.1" 404 None
    [2023-02-05, 14:55:37 IST] {authorization.py:137} ERROR - {"error":{"code":404,"message":"Schema is not present","errors":[{"domain":"global","reason":"notFound","message":"Schema is not present"}]}}
    [2023-02-05, 14:55:37 IST] {validate_schema.py:170} ERROR - Error on getting schema of kind 'osdu:wks:work-product-component--WellLog:2.2.0'
    [2023-02-05, 14:55:37 IST] {validate_schema.py:171} ERROR - 404 Client Error: Not Found for url: https://contoso.energy.azure.com/api/schema-service/v1/schema/osdu:wks:work-product-component--WellLog:2.2.0
    [2023-02-05, 14:55:37 IST] {validate_schema.py:314} WARNING - osdu:wks:work-product-component--WellLog:2.2.0 is not present in Schema service.
    [2023-02-05, 15:01:23 IST] {validate_schema.py:322} ERROR - Schema validation error. Data field.
    [2023-02-05, 15:01:23 IST] {validate_schema.py:323} ERROR - Manifest kind: osdu:wks:work-product-component--WellLog:1.1.0
    [2023-02-05, 15:01:23 IST] {validate_schema.py:324} ERROR - Error: 'string-value' is not of type 'number'
    
    Failed validating 'type' in schema['properties']['data']['allOf'][3]['properties']['SamplingStop']:
        {'description': 'The stop value/last value of the ReferenceCurveID, '
                        'typically the end depth of the logging.',
         'example': 7500,
         'title': 'Sampling Stop',
         'type': 'number',
         'x-osdu-frame-of-reference': 'UOM'}
    
    On instance['data']['SamplingStop']:
        'string-value'

실패한 참조 검사

참조 검사가 실패했기 때문에 레코드가 수집되지 않았습니다.

가능한 이유

참조된 레코드를 찾을 수 없습니다.
부모 레코드를 찾을 수 없습니다.
검색 서비스에서 5xx 오류를 throw합니다.

워크플로 상태

워크플로 상태가 finished로 표시됩니다. 잘못된 엔터티를 건너뛰고 수집을 계속하기 때문에 워크플로 상태 오류가 표시되지 않습니다.

솔루션

provide_manifest_integrity_task 또는 process_manifest_task에 대한 Airflow 작업 로그를 확인합니다.

샘플 Kusto 쿼리:

    OEPAirFlowTask
        | where DagName has "Osdu_ingest"
        | where DagTaskName == "provide_manifest_integrity_task" or DagTaskName has "process_manifest_task"
        | where Content has 'Search query "'or Content has 'response ids: ['
        | where RunID has "<run_id>"

참조 무결성 작업에 대한 오류 로그가 없으므로 디버그 로그 문을 검사 모든 외부 레코드가 검색 서비스를 통해 페치되었는지 확인합니다.

예를 들어 다음 샘플 추적 출력은 참조 무결성을 위해 검색 서비스를 통해 쿼리된 레코드를 보여줍니다.

    [2023-02-05, 19:14:40 IST] {search_record_ids.py:75} DEBUG - Search query "contoso-dp1:work-product-component--WellLog:5ab388ae0e140838c297f0e6559" OR "contoso-dp1:work-product-component--WellLog:5ab388ae0e1b40838c297f0e6559" OR "contoso-dp1:work-product-component--WellLog:5ab388ae0e1b40838c297f0e6559758a"

출력에는 검색되어 시스템에 있던 레코드가 표시됩니다. 레코드를 참조하는 관련 매니페스트 개체는 삭제되고 일부 레코드가 없는 경우 더 이상 수집되지 않습니다.

    [2023-02-05, 19:14:40 IST] {search_record_ids.py:141} DEBUG - response ids: ['contoso-dp1:work-product-component--WellLog:5ab388ae0e1b40838c297f0e6559758a:1675590506723615', 'contoso-dp1:work-product-component--WellLog:5ab388ae0e1b40838c297f0e6559758a    ']

매니페스트의 잘못된 법적 태그 또는 ACL

매니페스트에 잘못된 법적 태그 또는 ACL(액세스 제어 목록)이 포함되어 있으므로 레코드가 수집되지 않았습니다.

가능한 이유

ACL이 잘못되었습니다.
법적 태그가 잘못되었습니다.
스토리지 서비스에서 5xx 오류를 throw합니다.

워크플로 상태

워크플로 상태가 finished로 표시됩니다. 워크플로 상태 오류가 표시되지 않습니다.

솔루션

process_single_manifest_file_task 또는 process_manifest_task에 대한 Airflow 작업 로그를 확인합니다.

샘플 Kusto 쿼리:

    OEPAirFlowTask
    | where DagName has "Osdu_ingest"
    | where DagTaskName == "process_single_manifest_file_task" or DagTaskName has "process_manifest_task"
    | where LogLevel == "ERROR"
    | where RunID has "<run_id>"
    | order by ['time'] asc

샘플 추적 출력:

    "PUT /api/storage/v2/records HTTP/1.1" 400 None
    [2023-02-05, 16:57:05 IST] {authorization.py:137} ERROR - {"code":400,"reason":"Invalid legal tags","message":"Invalid legal tags: contoso-dp1-R3FullManifest-Legal-Tag-Test779759112"}

출력은 검색된 레코드를 나타냅니다. 누락된 검색 레코드에 해당하는 매니페스트 엔터티 레코드는 삭제되고 수집되지 않습니다.

    "PUT /api/storage/v2/records HTTP/1.1" 400 None
    [2023-02-05, 16:58:46 IST] {authorization.py:137} ERROR - {"code":400,"reason":"Validation error.","message":"createOrUpdateRecords.records[0].acl: Invalid group name 'data1.default.viewers@contoso-dp1.dataservices.energy'"}
    [2023-02-05, 16:58:46 IST] {single_manifest_processor.py:83} WARNING - Can't process entity SRN: surrogate-key:0ef20853-f26a-456f-b874-3f2f5f35b6fb

알려진 문제

참조 무결성 작업에 대한 특정 오류 로그가 없으므로 수동으로 디버그 로그 문을 검색하여 검색 서비스를 통해 모든 외부 레코드가 검색되었는지 확인해야 합니다.

다음 단계

매니페스트 수집 자습서로 이동하여 매니페스트 기반 파일 수집을 수행하는 방법을 알아봅니다.

자습서: 매니페스트 기반 파일 수집을 수행하기 위한 샘플 단계

다음을 통해 공유

Airflow 작업 로그를 사용하여 매니페스트 수집 문제 해결

매니페스트 수집 DAG 워크플로 유형

단일 매니페스트

일괄 업로드

필수 조건

실패한 DAG 실행

가능한 이유

워크플로 상태

솔루션

실패한 스키마 유효성 검사

가능한 이유

워크플로 상태

솔루션

실패한 참조 검사

가능한 이유

워크플로 상태

솔루션

매니페스트의 잘못된 법적 태그 또는 ACL

가능한 이유

워크플로 상태

솔루션

알려진 문제

다음 단계

참조

피드백

추가 리소스