파이썬의 50 라인 미만으로 데이터 청소 및 검증 파이프 라인 구축

저자의 이미지 | 표의 문자

데이터는 지저분합니다. 따라서 API에서 정보를 가져 오거나 실제 데이터 세트 등을 분석 할 때는 필연적으로 복제, 결 측값 및 유효하지 않은 항목으로 실행됩니다. 동일한 청소 코드를 반복적으로 작성하는 대신 잘 설계된 파이프 라인은 시간을 절약하고 데이터 과학 프로젝트에서 일관성을 보장합니다.

이 기사에서는 고정 된 내용에 대한 자세한 피드백을 제공하면서 일반적인 데이터 품질 문제를 처리하는 재사용 가능한 데이터 청소 및 검증 파이프 라인을 구축 할 것입니다. 결국, 데이터 세트를 정리하고 몇 줄의 코드로 비즈니스 규칙에 대해 검증 할 수있는 도구가 있습니다.

🔗 Github의 코드 링크

왜 데이터 청소 파이프 라인?

제조의 어셈블리 라인과 같은 데이터 파이프 라인을 생각해보십시오. 각 단계는 특정 기능을 수행하고 한 단계의 출력은 다음 단계의 입력이됩니다. 이 접근법은 코드가 다른 프로젝트에서보다 유지 관리 가능하고 테스트 가능하며 재사용 할 수 있도록합니다.

간단한 데이터 청소 파이프 라인
저자의 이미지 | diagrams.net (draw.io)

파이프 라인은 세 가지 핵심 책임을 처리합니다.

• 청소 : 복제를 제거하고 결 측값을 처리합니다 (이를 시작점으로 사용하십시오. 필요에 따라 많은 청소 단계를 추가 할 수 있습니다.)
• 검증 : 데이터가 비즈니스 규칙 및 제약을 충족하는지 확인하십시오
• 보고 : 처리 중에 어떤 변화가 있었는지 추적하십시오

개발 환경 설정

최근 버전의 Python을 사용하고 있는지 확인하십시오. 로컬로 사용하는 경우 가상 환경을 만들고 필요한 패키지를 설치하십시오.

원하는 경우 Google Colab 또는 유사한 노트북 환경을 사용할 수도 있습니다.

유효성 검사 스키마 정의

데이터를 검증하기 전에 “유효한”모습을 정의해야합니다. 유형 힌트를 사용하여 데이터 유형을 검증하는 파이썬 라이브러리 인 Pydantic을 사용하겠습니다.

class DataValidator(BaseModel):
    name: str
    age: Optional[int] = None
    email: Optional[str] = None
    salary: Optional[float] = None
    
    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v is not None and (v  100):
            raise ValueError('Age must be between 0 and 100')
        return v
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if v and '@' not in v:
            raise ValueError('Invalid email format')
        return v

이 스키마는 Pydantic의 구문을 사용하여 예상 데이터를 모델링합니다. 사용하려면 @field_validator 데코레이터, 당신은 필요합니다 @classmethod 데코레이터. 검증 논리는 나이가 합리적인 범위에 속하고 이메일에는 ‘@’기호가 포함되어 있습니다.

파이프 라인 클래스 구축

우리의 주요 파이프 라인 클래스는 모든 청소 및 검증 논리를 캡슐화합니다.

class DataPipeline:
    def __init__(self):
        self.cleaning_stats = {'duplicates_removed': 0, 'nulls_handled': 0, 'validation_errors': 0}

생성자는 통계 사전을 초기화하여 처리 중에 변경된 변경 사항을 추적합니다. 이를 통해 데이터 품질을 면밀히 살펴보고 시간이 지남에 따라 적용되는 청소 단계를 추적합니다.

데이터 청소 논리 작성

추가하자 clean_data 결 측값 및 중복 레코드와 같은 일반적인 데이터 품질 문제를 처리하는 방법 :

def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
    initial_rows = len(df)
    
    # Remove duplicates
    df = df.drop_duplicates()
    self.cleaning_stats['duplicates_removed'] = initial_rows - len(df)
    
    # Handle missing values
    numeric_columns = df.select_dtypes(include=[np.number]).columns
    df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].median())
    
    string_columns = df.select_dtypes(include=['object']).columns
    df[string_columns] = df[string_columns].fillna('Unknown')

이 접근법은 다양한 데이터 유형을 처리하는 데 현명합니다. 숫자 결 측값은 중간 값으로 채워집니다 (특이 치에 대한 평균보다 강력한), 텍스트 열은 자리 표시 자 값을 얻습니다. 중간 계산을 왜곡하지 않기 위해 먼저 중복 제거가 발생합니다.

오류 추적으로 유효성 검사 추가

유효성 검사 단계는 각 행을 개별적으로 처리하여 유효한 데이터와 자세한 오류 정보를 모두 수집합니다.

def validate_data(self, df: pd.DataFrame) -> pd.DataFrame:
    valid_rows = []
    errors = []
    
    for idx, row in df.iterrows():
        try:
            validated_row = DataValidator(**row.to_dict())
            valid_rows.append(validated_row.model_dump())
        except ValidationError as e:
            errors.append({'row': idx, 'errors': str(e)})
    
    self.cleaning_stats['validation_errors'] = len(errors)
    return pd.DataFrame(valid_rows), errors

이 행 바로 접근 방식은 하나의 나쁜 레코드가 전체 파이프 라인에 충돌하지 않도록합니다. 유효한 행은 프로세스를 통해 계속되고 오류는 검토를 위해 캡처됩니다. 이것은 문제를 깃발하면서 가능한 것을 처리 해야하는 생산 환경에서 중요합니다.

파이프 라인을 조정합니다

그만큼 process 메소드는 모든 것을 함께 묶습니다.

def process(self, df: pd.DataFrame) -> Dict[str, Any]:
    cleaned_df = self.clean_data(df.copy())
    validated_df, validation_errors = self.validate_data(cleaned_df)
    
    return {
        'cleaned_data': validated_df,
        'validation_errors': validation_errors,
        'stats': self.cleaning_stats
    }

반환 값은 정리 된 데이터, 유효성 검사 오류 및 처리 통계를 포함하는 포괄적 인 보고서입니다.

모든 것을 함께 모으십시오

실제로 파이프 라인을 사용하는 방법은 다음과 같습니다.

# Create sample messy data
sample_data = pd.DataFrame({
    'name': ['Tara Jamison', 'Jane Smith', 'Lucy Lee', None, 'Clara Clark','Jane Smith'],
    'age': [25, -5, 25, 35, 150,-5],
    'email': ['[email protected]', 'invalid-email', '[email protected]', '[email protected]', '[email protected]','invalid-email'],
    'salary': [50000, 60000, 50000, None, 75000,60000]
})

pipeline = DataPipeline()
result = pipeline.process(sample_data)

파이프 라인은 중복 레코드를 자동으로 제거하고, ‘알 수없는’으로 채워서 누락 된 이름을 처리하고, 누락 된 급여를 중간 값으로 채우고, 음의 연령 및 유효하지 않은 이메일에 대한 유효성 검사 오류를 플래그합니다.

🔗 Github에서 전체 스크립트를 찾을 수 있습니다.

파이프 라인을 확장합니다

이 파이프 라인은 당신이 구축 할 수있는 기초 역할을합니다. 특정 요구에 대한 이러한 개선 사항을 고려하십시오.

맞춤형 청소 규칙: 전화 번호 또는 주소 표준화와 같은 도메인 별 청소 방법을 추가하십시오.

구성 가능한 유효성 검사: 동일한 파이프 라인이 다른 데이터 유형을 처리 할 수 ​​있도록 Pydantic 스키마를 구성 할 수 있도록합니다.

고급 오류 처리: 일시적 오류 또는 일반적인 실수에 대한 자동 보정에 대한 재시 도로 로직을 구현하십시오.

성능 최적화: 대형 데이터 세트의 경우 벡터화 된 작업 또는 병렬 처리를 고려하십시오.

마무리

데이터 파이프 라인은 개별 데이터 세트를 청소하는 것이 아닙니다. 그들은 신뢰할 수 있고 유지 관리 가능한 시스템을 구축하는 것에 관한 것입니다.

이 파이프 라인 접근 방식은 프로젝트의 일관성을 보장하고 요구 사항이 변경됨에 따라 비즈니스 규칙을 쉽게 조정할 수 있도록합니다. 이 기본 파이프 라인부터 시작한 다음 특정 요구에 맞게 사용자 정의하십시오.

열쇠는 평범한 작업을 처리하는 신뢰할 수 있고 재사용 가능한 시스템을 갖추고있어 깨끗한 데이터에서 통찰력을 추출하는 데 집중할 수 있습니다. 행복한 데이터 청소!

발라 프리 야 c 인도의 개발자이자 기술 작가입니다. 그녀는 수학, 프로그래밍, 데이터 과학 및 컨텐츠 제작의 교차점에서 일하는 것을 좋아합니다. 그녀의 관심 분야와 전문 지식에는 DevOps, 데이터 과학 및 자연어 처리가 포함됩니다. 그녀는 독서, 쓰기, 코딩 및 커피를 즐깁니다! 현재 그녀는 자습서, 방법 안내, 의견 조각 등을 통해 개발자 커뮤니티와 지식을 배우고 공유하는 작업을하고 있습니다. Bala는 또한 매력적인 리소스 개요 및 코딩 자습서를 만듭니다.