위대한 기대 강좌: Python을 사용한 데이터 유효성 검사

Great Expectations 설정하기

이 튜토리얼에서는 GX Core, Great Expectations의 오픈 소스 버전을 사용하여 Pandas DataFrame을 검증하는 방법을 배우게 됩니다. 컨텍스트 설정, Pandas 데이터 소스 등록, 기대값 정의 및 데이터 배치 유효성 검사 과정을 안내합니다.

참고: 데이터랩 노트북을 사용하여 따라하시는 것을 권장하지만, 자체 Python 스크립트를 만들 수도 있습니다.

1. Great Expectations 설치

전제 조건

• Python 3.9에서 3.12가 설치되어 있어야 합니다.
• 충돌을 피하기 위해 Great Expectations을 가상 환경 안에 설치하는 것이 매우 권장됩니다 (면책: 가상 환경 설정은 이 문서의 범위를 벗어납니다).
• 샘플 데이터셋.

참고: 제공된 DataLab 노트북을 사용하는 경우 이러한 전제 조건이 이미 충족되었습니다. 건너뛰어도 좋습니다.

다음 명령을 사용하여 pip를 통해 GX를 설치하십시오:

pip install great_expectations

이 명령은 핵심 패키지와 모든 필수 종속 항목을 설치합니다.

2. 데이터 컨텍스트 초기화

Great Expectations는 구성을 관리하기 위한 데이터 컨텍스트가 필요합니다. 구성을 영구적으로 유지하지 않도록 하기 위해 일시적 데이터 컨텍스트를 사용합니다.

import great_expectations as gx

# 일시적 데이터 컨텍스트 가져오기
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

첫 번째 데이터 유효성 검사 스위트 생성

GX가 설정되었으므로 데이터 유효성 검사 스위트를 생성해 봅시다.

1. 데이터 원본에 연결하고 데이터 자산 생성

데이터 원본은 Great Expectations를 데이터에 연결하고, 데이터 자산은 데이터의 특정 하위 집합(예: 테이블, DataFrame 또는 파일)을 나타냅니다.

이 경우에는 inventory_parts_df라는 DataFrame에 연결하기 위해 모든 준비를 할 것입니다. 샘플 데이터 세트는 제공된 DataLab에 있으며, SQL 블록을 실행하면 생성됩니다:

DataLab를 사용하지 않는 경우 샘플 데이터로 자체 DataFrame을 생성하십시오.

이제 데이터 소스와 자산을 만듭니다:

# 판다스 데이터 소스 추가
data_source = context.data_sources.add_pandas(name="inventory_parts")
# 데이터 소스에 데이터 자산 추가
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. 배치 정의 추가

배치 정의는 데이터를 확인하기 위해 식별하고 구성합니다. 여기서 전체 DataFrame을 포함하는 배치 정의를 추가합니다:

# Batch 정의 이름 정의
batch_definition_name = "inventory_parts_batch"
# Batch 정의 추가
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Batch 검색

Batch는 Batch 정의에 연결된 데이터 집합입니다. 데이터를 확인하려면 Batch를 검색하고 DataFrame에 연결해야 합니다. 이 경우 inventory_parts_df를 사용합니다:

# Batch 매개변수 정의
batch_parameters = {"dataframe": inventory_parts_df}
# Batch 검색
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. 스위트 생성 및 기대값 정의

기대값은 데이터 유효성을 검증하기 위한 규칙입니다. 이 예에서는 다음과 같은 간단한 기대값을 정의합니다:

1. inventory_id 값이 null이 아닌지 확인합니다.
2. part_num 값이 고유한지 확인합니다.

기대값 스위트 생성
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
기대값 추가
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
컨텍스트에 기대값 스위트 추가
context.suites.add(suite)

기대값 갤러리에서 사용 가능한 모든 기대값을 탐색할 수 있습니다. 몇 가지 더 추가하는 것을 권장합니다!

기대값을 정의한 후 GX는 기대값 스위트 구성을 출력합니다:

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

스위트에는 다음과 같은 세부 정보가 포함됩니다:

1. 스위트 이름 및 ID: 고유한 이름(inventory_parts_suite)과 스위트를 추적하고 관리하기 위한 식별자가 포함됩니다.
2. 기대 사항: 각 규칙은 다음을 지정합니다:
• 검사 유형(예: 열에 null 값이나 고유 항목이 없도록 보장).
• 검증할 열과 같은 매개변수.
• 각 기대에 대한 메타데이터 및 고유 ID, 더 쉬운 추적 및 사용자 정의를 가능하게 합니다.
3. 메타데이터: 도구와의 호환성을 보장하는 Great Expectations의 버전 정보.
4. 노트: 스위트에 대한 설명적인 코멘트를 추가하는 자리 (선택 사항).

이 구조화된 출력은 문서화뿐만 아니라 데이터셋을 검증하는 재사용 가능한 구성으로 작용하여 여러분의 기대치가 명확히 정의되고 추적 가능하며 미래 사용을 위해 준비됩니다.

5. 데이터 유효성 검사

마지막으로, 정의된 기대치에 대해 일괄 처리를 유효성 검사하고 결과를 평가합니다.

# 스위트에 대한 데이터 유효성 검사
validation_results = batch.validate(suite)
# 결과 평가
print(validation_results)

검증을 실행한 후 Great Expectations은 데이터셋이 정의된 기대치를 충족하는지에 대한 상세 보고서를 제공합니다:

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

이 보고서는 데이터의 품질을 상세히 설명하여 성공과 실패를 강조합니다. 결과의 간단한 설명은 다음과 같습니다:

전반적인 유효성 검사: 유효성 검사 결과가 부분적으로 성공했습니다: 기대치의 50%가 통과했고 50%가 실패했습니다. 실패한 기대치는 주의해야 할 데이터 품질 문제를 나타냅니다. 이 경우에는 하나의 열이 정의된 규칙을 충족하지 못했습니다.

기대치 1: inventory_id는 누락된 값이 없어야 합니다

• 결과: 통과됨
• 설명: inventory_id 열의 모든 값이 존재하며, null 또는 누락된 항목이 없습니다. 이는 해당 열에 대한 데이터 완전성이 좋다는 것을 나타냅니다.

기대 2: part_num은 고유한 값이어야 합니다

• 결과:실패
• 설명: part_num 열에는 97.98%의 중복 값이 포함되어 있어 몇 가지 값만이 고유합니다.
• 하이라이트:
• 예시 중복 값으로는 “3069b”와 “33291”이 있습니다.
• 이 도구는 또한 이러한 중복이 얼마나 자주 나타나는지와 그들의 행 위치를 보여주어 문제를 쉽게 찾고 수정할 수 있도록 도와줍니다.

물론, 이것은 단순한 샘플 데이터셋이며, 통과 및 실패하는 예상을 의도적으로 포함하여 두 가지 유효성 검사 결과를 확인할 수 있습니다.

이게 다입니다! 당신은 성공적으로 end-to-end 데이터 유효성을 실행했습니다.

데이터 파이프라인에 Great Expectations 통합

프로덕션 환경에서는 유효성 검사를 워크플로에 직접 통합하여 각 단계에서 데이터 품질을 계속 모니터링해야 합니다.

이 섹션에서는 Great Expectations을 데이터 파이프라인에 통합하는 방법에 대해 논의하겠습니다.

이것들은 아이디어를 제공하기 위한 예시이며, 여기에 포함되지 않은 추가 구성이 필요할 수 있습니다. 최신 구문에 대한 각 도구의 문서를 확인하세요!

ETL 도구와의 통합

인기있는 ETL 도구인 Apache Airflow나 Prefect와의 Great Expectations 통합은 비교적 간단합니다. 데이터 검증 단계를 ETL 프로세스에 직접 통합하면 데이터 문제를 실시간으로 파악하고 처리할 수 있어 하류 분석에 영향을 미치기 전에 해결할 수 있습니다.

자동화된 ETL 워크플로의 일부로 데이터 검증을 실행하기 위해 Great Expectations를 Prefect와 통합하는 간단한 예제를 살펴봅시다:

from prefect import task, Flow
import great_expectations as ge
# Great Expectations 검증 실행을 위한 작업 정의
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # 검증 결과 확인 및 검증 실패 시 경고 발생
    if not results["success"]:
        raise ValueError("Data validation failed!")
# ETL 플로우 정의
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# 플로우 실행
flow.run()

이 예제에서는 Prefect 플로우를 정의하여 Great Expectations 검증을 실행하는 작업을 포함시킵니다.

validate_data() 작업은 Great Expectations 컨텍스트를 로드하고 데이터 배치를 검색한 후 기대값 스위트를 적용합니다.

데이터가 유효성 검사 기준을 충족하지 않는 경우 작업은 경고를 발생시켜 워크플로를 중지하고 하류 오류를 방지합니다.

지속적인 데이터 유효성 검사

크론 작업 또는 Apache Airflow와 같은 관리형 서비스와 같은 다양한 도구를 사용하여 유효성 검사 작업을 예약할 수 있습니다. 이 예에서는 데이터 파이프라인을 조정하는 데 적합한 Airflow를 사용하여 유효성 검사 실행을 예약하는 방법을 보여줍니다.

매일 Great Expectations 유효성 검사를 실행하도록 Airflow DAG(방향성 비순환 그래프)를 설정하는 방법은 다음과 같습니다:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# DAG을 정의하고 일일 실행 일정 설정
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # 매일 한 번 실행
)
# 유효성 검사를 실행하는 함수 정의
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Airflow에서 작업 설정
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# DAG에서 작업 설정
validation_task

이 예에서는 일일로 검증 실행 일정을 설정한 DAG를 정의합니다 (@daily).

run_validation() 함수는 Great Expectations 컨텍스트를로드하고 정의된 예상 스위트를 데이터에 대해 실행하여 유효성을 검사합니다.

Great Expectations를 사용한 데이터 유효성을 위한 모범 사례

확장성과 효율성을 위해 모범 사례를 따르는 것이 좋으며, Great Expectations를 사용한 데이터 유효성 검사도 예외는 아닙니다.

작게 시작하고 반복하세요

기초 데이터 품질 확인부터 시작하여 서서히 확장하십시오. 처음에는 기본 기대치에 초점을 맞추는 것이 좋습니다. 이렇게 하면 프로세스를 과도하게 복잡하게 만들지 않아 통합이 원활하고 문제 해결이 쉬워집니다. 데이터셋에 대한 이해가 개선되면 보다 복잡한 유효성 검사를 추가할 수 있습니다.

팀 간 협업

데이터 품질은 단순히 기술적인 문제가 아닙니다. 비즈니스 팀 간 협력을 통해 기대치를 정의하고 구현된 유효성 검사가 근간이 되는 비즈니스 로직과 목표와 일치하는지 확인하십시오. 이 교차 기능적 접근 방식은 데이터가 의도된 목적에 부합하고 모든 이해관계자의 요구 사항을 충족시키는 것을 보장합니다.

가능한 경우 자동화

가능한 경우 프로세스를 자동화하여 데이터 유효성 검사를 데이터 파이프라인에 통합하십시오. 자동화된 유효성 검사를 통합하면 수동 개입 없이 데이터 품질을 계속 모니터링할 수 있어 효율성이 크게 향상됩니다.

결론

좋은 작업입니다! Great Expectations에서 데이터를 구성하고 유효성을 검사하는 방법을 배웠습니다. 이러한 기술은 워크플로우에서 높은 데이터 품질과 투명성을 유지하는 데 도움이 될 것입니다.

기술을 향상시키려면 다음 자료를 확인해보세요:

• 파이썬에서 ETL 및 ELT: 데이터를 효과적으로 변환하고 이동하는 방법을 배웁니다.
• 데이터 품질 소개: 데이터 품질 관리의 기본을 탐색합니다.
• 파이썬에서 데이터 정리: 정확성과 일관성을 보장하기 위한 데이터 정리 기술을 숙달합니다.
• 데이터 품질 차원 요약 시트: 데이터 품질 차원에 대한 편리한 안내서입니다.