Python UDF 로깅을 활용하여 파이프라인 모니터링 가능성을 확보하는 방법

*이 콘텐츠는 AI 번역 도구를 사용하여 번역되었습니다. 정확성을 기하기 위해 노력하고 있으나, 자동 번역에는 오류나 누락이 포함될 수 있습니다. 정확한 정보는 영문 원본을 참고해 주시기 바랍니다.

BigQuery의 Python 사용자 정의 함수(UDF)에서 새롭게 사용할 수 있는 로깅 기능을 소개합니다. 이제 개발자는 BigQuery UDF 내에서 Python의 표준 로깅 라이브러리를 직접 사용하여 구조화된 로그를 Cloud Logging으로 원활하게 내보낼 수 있습니다. 별도의 외부 플러그인이나 커스텀 커넥터는 필요하지 않습니다.

UDF 로깅은 광범위한 운영 모니터링 및 디버깅 워크플로를 지원하지만, 본 문서에서는 이 기능을 활용하여 내결함성과 복원력이 뛰어난 데이터 파이프라인을 구축하는 사례를 보여드리겠습니다. 아래 예시에서는 쿼리 실행 중에 행 단위로 진단 원격 분석 데이터를 캡처합니다. 이를 통해 팀은 전체 워크로드를 중단하지 않고도 SQL을 통해 문제가 발생한 레코드를 쉽게 찾아내고, 실시간 알림을 트리거하며, 문제를 분류할 수 있습니다.

본 게시물에서는 Python UDF 로깅을 시작하는 방법을 알아보고, 이커머스 이벤트 수집 파이프라인을 실질적인 엔드 투 엔드 예시로 살펴보겠습니다.

아키텍처 개요

워크플로는 세 단계로 구성됩니다.

1. 인라인 검증: Python UDF가 수신되는 JSON 페이로드를 파싱합니다. 유효한 레코드는 대상 테이블로 전달되고, 손상되었거나 규정을 준수하지 않는 행은 logging.error()를 트리거하고 NULL을 반환하여 쿼리 실패를 방지합니다.

2. 실시간 알림: 오류율이 SLA 기준점을 초과하면 Cloud Monitoring 로그 기반 측정항목을 통해 팀에 알림이 전송됩니다.

3. SQL 분류 및 재처리: 데이터 엔지니어는 연결된 _AllLogs 뷰를 통해 BigQuery에서 표준 SQL로 격리된 페이로드를 쿼리하고, 근본 원인을 진단한 후에 복구된 데이터를 다시 수집합니다.

본 시연에서는 데이터 검증 및 변환에 Python을 사용하지만, SQL 기반 변환을 수행하는 파이프라인에도 Python 기반 로깅 및 검증을 원활하게 통합할 수 있습니다.

빠른 시작 둘러보기: 단계별 안내

1단계: 일반 Python UDF 만들기

이 UDF는 JSON을 파싱하고 필수 필드가 있는지 확인합니다. 파싱에 실패하거나 필수 키가 누락된 경우 구조화된 진단 JSON을 로깅하고 NULL을 반환합니다.

2단계: 로깅을 사용 설정한 상태로 수집 실행하기

이 예시에서는 수신되는 사용자 작업(클릭, 조회, 결제)을 처리하는 이커머스 이벤트 수집 파이프라인을 시뮬레이션합니다. 목표는 유효한 이벤트를 추출하여 정제된 분석 테이블에 저장하는 동시에 손상되거나 스키마가 누락된 페이로드를 적절하게 격리하는 것입니다.

SET @@udf.properties를 사용하여 세션에서 로깅을 사용 설정하세요.

수집 결과

쿼리는 중단 없이 원활하게 실행되며 문제가 있는 행(2 및 3)은 Cloud Logging으로 라우팅됩니다.

3단계: Cloud Logging에서 실시간 알림 설정하기

잘못된 형식의 페이로드가 허용 가능한 기준점을 초과하는 즉시 사고 긴급 대응팀에 알리도록 알림을 구성할 수 있습니다.

Google Cloud 콘솔에서 설정(원클릭)

1. Google Cloud 콘솔에서 로그 탐색기를 엽니다.

2. UDF 루틴에 대한 로그 필터를 붙여넣습니다.

3. 상단 작업 모음에서 알림 만들기를 클릭합니다.

4. 알림 채널(예: Slack, PagerDuty, 이메일)을 설정하고 정책을 저장합니다.

4단계: BigQuery 로그 분석으로 문제가 있는 행 분류하기

"참고: BigQuery SQL을 통해 로그를 쿼리하려면 Cloud Logging 버킷에서 BigQuery 로그 분석을 사용 설정하고 연결된 데이터 세트를 만들어야 합니다."

연결된 _AllLogs 뷰를 통해 BigQuery SQL을 사용하여 격리된 페이로드를 쿼리하세요.

로그 분석 출력

데이터 복구

전체 원시 페이로드가 Cloud Logging에 보존되므로 업스트림 데이터가 수정되면 SQL을 사용하여 격리된 레코드를 직접 복구할 수 있습니다.

최종 clean_events 테이블 결과:

이제 모든 유효한 데이터와 복구된 데이터가 데이터 손실 없이 프로덕션 테이블에 통합됩니다.

기타 사용 사례

• 섀도 실행 및 안전한 마이그레이션: UDF에서 기존 비즈니스 로직과 새로운 비즈니스 로직을 나란히 실행합니다. 기존 결과를 프로덕션 환경에 반환하는 동시에 계산 결과 차이를 Cloud Logging에 로깅합니다.

• 개인 식별 정보 수정 감사: BigQuery에 기록하기 전에 민감한 토큰(예: 사회보장번호(SSN) 또는 계좌 번호)을 마스킹하고 익명화된 감사 이벤트를 Cloud Logging으로 내보냅니다.

• ML 추론 드리프트: 내장된 점수 로직에서 어휘에 없는 입력 또는 낮은 모델 신뢰도가 발견되면 경고를 로깅합니다.

권장사항

• 디버그 출력 사용 설정: 로그를 캡처하려는 세션에 SET @@udf.properties = JSON '{"enableDebugOutput": true}';를 포함하세요.

• 로그에 민감한 정보 포함 금지: Cloud Logging은 로깅 뷰어 권한이 있는 사용자에게 표시됩니다. 로그인 사용자 인증 정보나 보안 비밀은 로깅하지 마세요.

• 비용 관리: 처리량이 많은 파이프라인에서는 선택적으로 로깅하여 로깅 수집 볼륨을 관리하세요. 예를 들어, 오류가 발생한 경우에만 로깅하거나 샘플링을 적용할 수 있습니다.

시작하기

• BigQuery Python UDF 문서

• BigQuery 로그 분석 개요