장기 바로 가기 : 야심 찬 데이터 엔지니어를위한 자동화 된 워크 플로

저자의 이미지 | 표의 문자

틀 소개

데이터 엔지니어로서의 근무일에 몇 시간이 걸리며 이미 일상적인 작업에 익사하고 있습니다. CSV 파일은 유효성 검사가 필요하며 데이터베이스 스키마에는 업데이트가 필요하고 데이터 품질 검사가 진행 중이며 이해 관계자는 어제 (및 그 전날)와 동일한 보고서를 요청하고 있습니다. 친숙한 소리?

이 기사에서는 시간 소비 수동 데이터 엔지니어링 작업을 SET-IT-and-Forget-IT 시스템으로 변환하는 실제 자동화 워크 플로우를 살펴 보겠습니다. 우리는 구현하는 데 몇 달이 걸리는 복잡한 엔터프라이즈 솔루션에 대해 이야기하지 않습니다. 이들은 즉시 사용할 수있는 간단하고 유용한 스크립트입니다.

메모: 기사의 코드 스 니펫은 스크립트의 클래스를 사용하는 방법을 보여줍니다. 전체 구현은 기럽 필요에 따라 사용하고 수정할 수있는 저장소. 🔗 코드에 대한 github 링크

틀 “간단한”데이터 엔지니어링 작업의 숨겨진 복잡성

솔루션으로 뛰어 들기 전에 간단한 데이터 엔지니어링 작업이 시간 싱크가되는 이유를 이해해 봅시다.

// 데이터 유효성 검사는 숫자를 확인하는 것이 아닙니다

새 데이터 세트를 받으면 숫자가 숫자임을 확인하는 것 이상의 유효성이 있습니다. 확인해야합니다.

• 시간 동안 스키마 일관성
• 다운 스트림 프로세스가 파손될 수있는 데이터 드리프트
• 기술 검증에 의해 잡히지 않은 비즈니스 규칙 위반
• 실제 데이터로 만 표면하는 가장자리 케이스

// 파이프 라인 모니터링에는 일정한 경계가 필요합니다

데이터 파이프 라인은 창의적인 방식으로 실패합니다. 성공적인 실행은 올바른 출력을 보장하지 않으며 실패한 실행이 항상 명백한 경고를 유발하지는 않습니다. 수동 모니터링은 다음과 같습니다.

• 여러 시스템에서 로그를 확인합니다
• 실패를 외부 요인과 상관시킵니다
• 각 오류의 다운 스트림 영향을 이해합니다
• 종속 프로세스에서 복구 조정

// 보고서 생성에는 쿼리 이상이 포함됩니다

자동화 된보고는 다음과 같은 요소까지 간단하게 들립니다.

• 동적 날짜 범위 및 매개 변수
• 데이터 값을 기반으로 한 조건부 형식
• 액세스 수준이 다른 다른 이해 관계자에게 배포됩니다
• 누락 된 데이터 및 에지 케이스 처리
• 보고서 템플릿에 대한 버전 제어

복잡성은 이러한 작업이 다른 환경에서 규모로 안정적으로 발생해야 할 때 배가됩니다.

틀 워크 플로 1 : 자동 데이터 품질 모니터링

어제의 데이터로드가 성공적으로 완료되었는지 수동으로 매일 첫 시간을 보낼 것입니다. 동일한 쿼리를 실행하고 동일한 메트릭을 확인하고 다른 사람이 읽지 않은 스프레드 시트에 동일한 문제를 문서화합니다.

// 해결책

파이썬으로 워크 플로우를 작성 하여이 매일 집안일을 배경 프로세스로 변환하고 그렇게 사용할 수 있습니다.

from data_quality_monitoring import DataQualityMonitor
# Define quality rules
rules = [
    {"table": "users", "rule_type": "volume", "min_rows": 1000},
    {"table": "events", "rule_type": "freshness", "column": "created_at", "max_hours": 2}
]

monitor = DataQualityMonitor('database.db', rules)
results = monitor.run_daily_checks()  # Runs all validations + generates report

// 스크립트의 작동 방식

이 코드는 데이터 테이블의 품질 검사관처럼 작동하는 스마트 모니터링 시스템을 만듭니다. 초기화 할 때 DataQualityMonitor 클래스는 모든 품질 규칙을 포함하는 구성 파일을로드합니다. 시스템에서 데이터를 “좋은”만드는 점검 목록으로 생각하십시오.

그만큼 run_daily_checks 방법은 데이터베이스의 각 테이블을 거치고 유효성 검사 테스트를 실행하는 기본 엔진입니다. 테이블이 품질 테스트에 실패하면 시스템은 자동으로 올바른 사람에게 알림을 보냅니다. 따라서 더 큰 문제를 일으키기 전에 문제를 해결할 수 있습니다.

그만큼 validate_table 메소드는 실제 점검을 처리합니다. 데이터 볼륨을 검토하여 레코드가 누락되지 않았는지 확인하고 정보가 최신 상태인지 확인하고, 결 측값을 잡을 수있는 완전성을 확인하고, 테이블 간의 관계가 여전히 의미가 있는지 확인하기 위해 일관성을 검증합니다.

▶ ️ 데이터 품질 모니터링 스크립트를 받으십시오

틀 워크 플로 2 : 동적 파이프 라인 오케스트레이션

기존의 파이프 라인 관리는 실행을 지속적으로 모니터링하고, 상황이 실패했을 때 수동으로 재실행을 유발하며, 다음 작업을 시작하기 전에 어떤 의존성을 확인하고 업데이트 해야하는지 기억하려고합니다. 반응성이고 오류가 발생하며 확장되지 않습니다.

// 해결책

변화하는 조건에 적응하고 그렇게 사용할 수있는 스마트 오케스트레이션 스크립트 :

from pipeline_orchestrator import SmartOrchestrator

orchestrator = SmartOrchestrator()

# Register pipelines with dependencies
orchestrator.register_pipeline("extract", extract_data_func)
orchestrator.register_pipeline("transform", transform_func, dependencies=["extract"])
orchestrator.register_pipeline("load", load_func, dependencies=["transform"])

orchestrator.start()
orchestrator.schedule_pipeline("extract")  # Triggers entire chain

// 스크립트의 작동 방식

그만큼 SmartOrchestrator 클래스는 모든 파이프 라인 의존성 맵을 구축하여 시작하므로 다른 사람이 시작하기 전에 어떤 작업을 완료 해야하는지 알 수 있습니다.

파이프 라인을 실행하고 싶을 때 schedule_pipeline 방법 먼저 모든 전제 조건 조건이 충족되는지 확인합니다 (필요한 데이터를 사용할 수 있고 신선한 지 확인). 모든 것이 좋아 보인 경우 현재 시스템 부하 및 데이터 볼륨을 고려하여 작업을 실행하는 가장 좋은 방법을 결정하는 최적화 된 실행 계획을 만듭니다.

그만큼 handle_failure 방법은 어떤 유형의 고장이 발생했는지 분석하고 그에 따라 응답하는데, 이는 간단한 재 시도, 데이터 품질 문제 조사 또는 문제가 수동으로주의가 필요할 때 인간에게 경고하는지 여부에 따라 응답합니다.

▶ ️ 파이프 라인 오케스트레이터 스크립트를 받으십시오

틀 워크 플로 3 : 자동 보고서 생성

데이터에서 일하는 경우 인간 보고서 생성기가 될 수 있습니다. 매일 매일 “빠른 보고서”에 대한 요청을 가져 오는 데 1 시간이 걸리며 다음 주에 약간 다른 매개 변수로 다시 요청할 것입니다. 실제 엔지니어링 작업은 임시 분석 요청을 위해 제쳐두고 있습니다.

// 해결책

자연 언어 요청에 따라 보고서를 생성하는 자동보고 생성기 :

from report_generator import AutoReportGenerator

generator = AutoReportGenerator('data.db')

# Natural language queries
reports = [
    generator.handle_request("Show me sales by region for last week"),
    generator.handle_request("User engagement metrics yesterday"),
    generator.handle_request("Compare revenue month over month")
]

// 스크립트의 작동 방식

이 시스템은 평범한 영어 요청을 잠들지 않고 이해하는 데이터 분석가 조수를 갖는 것처럼 작동합니다. 누군가가 보고서를 요청하면 AutoReportGenerator 먼저 NLP (Natural Language Processing)를 사용하여 판매 데이터, 사용자 지표 또는 성능 비교를 요구하는지 여부에 관계없이 원하는 것을 정확하게 파악합니다. 그런 다음 시스템은 보고서 템플릿 라이브러리를 검색하여 요청과 일치하는 템플릿을 찾거나 필요한 경우 새 템플릿을 만듭니다.

요청을 이해하면 올바른 데이터를 효율적으로 가져오고 해당 쿼리를 실행하며 결과를 전문적인 보고서로 형식화하는 최적화 된 데이터베이스 쿼리를 구축합니다. 그만큼 handle_request 메소드는 모든 것을 함께 연결하고 “지난 분기에 대한 지역별 판매 표시”또는 “매일 활성 사용자가 10%이상 떨어질 때 경고”와 같은 요청을 처리 할 수 ​​있습니다.

▶ ️ 자동 보고서 생성기 스크립트를 가져옵니다

틀 자신을 압도하지 않고 시작합니다

// 1 단계 : 가장 큰 고통 지점을 선택하십시오

한 번에 모든 것을 자동화하려고하지 마십시오. 워크 플로에서 가장 시간이 많이 걸리는 수동 작업을 식별하십시오. 일반적으로 이것은 다음 중 하나입니다.

• 일일 데이터 품질 검사
• 수동 보고서 생성
• 파이프 라인 고장 조사

이 하나의 작업에 대한 기본 자동화로 시작하십시오. 사례의 70%를 처리하는 간단한 스크립트조차도 상당한 시간을 절약 할 수 있습니다.

// 2 단계 : 모니터링 및 경고 구축

첫 번째 자동화가 실행되면 지능형 모니터링을 추가하십시오.

• 성공/실패 알림
• 성능 메트릭 추적
• 인간 에스컬레이션으로 예외 처리

// 3 단계 : 커버리지 확장

첫 번째 자동화 된 워크 플로가 효과적이면 다음으로 가장 큰 시간 싱크대를 식별하고 유사한 원칙을 적용하십시오.

// 4 단계 : 점을 연결하십시오

자동화 된 워크 플로 연결을 시작하십시오. 데이터 품질 시스템은 파이프 라인 오케스트레이터에게 알려야합니다. 오케스트레이터는 보고서 생성을 트리거해야합니다. 각 시스템은 통합 될 때 더 가치가 있습니다.

틀 일반적인 함정과 피하는 방법

// 첫 번째 버전을 과도하게 엔지니어링합니다

트랩 : 무엇이든 배포하기 전에 모든 에지 케이스를 처리하는 포괄적 인 시스템 구축.
수정 : 80% 사례로 시작하십시오. 대부분의 시나리오에 맞는 것을 배포 한 다음 반복하십시오.

// 오류 처리를 무시합니다

트랩 : 자동화 된 워크 플로가 항상 완벽하게 작동한다고 가정합니다.
수정 : 첫날부터 모니터링 및 경고를 구축하십시오. 실패를 계획하고, 그들이 일어나지 않기를 바라지 마십시오.

// 이해하지 않고 자동화

트랩 : 먼저 고정되는 수동 프로세스를 자동화하는 대신 먼저 고정합니다.
수정 : 자동화하기 전에 수동 프로세스를 문서화하고 최적화하십시오.

틀 결론

이 기사의 예는 Python 표준 라이브러리 만 사용하여 실시간 절약 및 품질 개선을 나타냅니다.

작게 시작하십시오. 하루의 30 분 이상 소비되는 워크 플로를 선택하고 이번 주에 자동화하십시오. 영향을 측정하십시오. 작동하는 것과 그렇지 않은 것으로부터 배우십시오. 그런 다음 자동화를 다음으로 가장 큰 시간 싱크대로 확장하십시오.

최고의 데이터 엔지니어는 데이터 처리에 능숙하지 않습니다. 그들은 지속적인 개입없이 데이터를 처리하는 시스템을 구축하는 데 능숙합니다. 이것이 데이터 엔지니어링 작업과 진정한 엔지니어링 데이터 시스템의 차이점입니다.

먼저 무엇을 자동화 하시겠습니까? 의견에 알려주십시오!

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