웹 스크래핑과 Google 서비스에 데이터 적재를 지원하는 프레임워크
- 이커머스 데이터를 수집, 변환, 적재하는 과정을 반복하면서 어떠한 작업 패턴이 있음을 파악하고 패턴을 구조화하여 코드를 간소화하고 싶었습니다.
- 설계 당시에는 작은 패턴을 계속해서 구조화하며 전체적인 흐름을 만들어냈고, 후행적으로 아래 그림 1-1과 같은 흐름도로 표현할 수 있습니다.
- 컨텍스트는 사용자 입력과 실행 정보를 담은 딕셔너리 집합으로, ETL 프로세스 전과정에서 메서드 또는 객체 간에 전달됩니다.
- 모든 쿼리에 대해 반복적으로 HTTP 요청 및 파싱을 수행합니다.
- List[Dict] 또는 pandas DataFrame 형식의 가공된 데이터는 하나의 객체로 취합되어 빅쿼리 또는 구글 시트에 적재됩니다.
- 환경변수
- 사용자 입력과 중간 결과는 "컨텍스트"라는 딕셔너리 객체에 들어가며, ETL 프로세스 전과정에서 메서드 또는 객체 간에 전달됩니다.
- 입력
- ETL 프로세스의 입력에는 쿼리 배열(키워드, 상품코드 등) 또는 쿼리 범위(시작일, 종료일 등)가 주어집니다.
- 여기서 쿼리는 HTTP GET 요청 시 Parameter 또는 POST 요청 시 Body에 들어가는 값을 의미합니다.
- 추출
- 비동기적으로 HTTP 요청하는 것이 시간 효율적이지만, 도메인 또는 경로에 따라 동시 요청 제한이 있을 수 있습니다.
- 쿼리 범위가 있다면 배열로 변환하고, 쿼리 배열에 대해 동기적/비동기적으로 반복문을 수행합니다.
- 변환
- 모든 가공된 데이터는 적재를 위해 pandas의 DataFrame 객체로 변환될 수 있어야 합니다.
- 따라서, 결과는 깊이가 1인 딕셔너리의 배열 객체 또는 DataFrame 객체가 됩니다.
- 적재할 테이블의 열(키)과 응답 결과에서 대응되는 값을 1대1로 맵핑하는 과정이 포함됩니다.
- 적재
- 모든 데이터는 빅쿼리 또는 구글 시트에 적재됩니다.
- 추가하기(append), 덮어쓰기(replace), 중복없이 추가하기(ignore), 중복없이 덮어쓰기(upsert) 방식이 있습니다.
- 알림
- 사내 메신저인 네이버 웍스에 Bot API를 사용해 결과(데이터 길이, 업로드 처리 상태 등)를 알림으로 보냅니다.
- 직접 정의한 템플릿 문법에 따라 메시지 내용에 텍스트 포맷팅으로 값을 동적 할당합니다.
- 로깅
- 오류 처리를 위해 입력, 추출, 변환, 적재, 알림 과정의 전후로 INFO 또는 DEBUG 수준의 로그를 남깁니다.
- 체크포인트를 남기고 특정 체크포인트에서 중간 저장 또는 강제 종료시키는 기능을 추가한다면 효과적인 디버깅이 가능합니다.
- 그림 1-1의 ETL 흐름도 및 주요 패턴을 Spider라는 웹 스크래퍼 클래스에서 구체적인 로직을 구현합니다.
- Spider의 메서드 실행 순서를 더 알아보기 쉽게 그림 1-2와 같은 다이어그램으로 나타낼 수 있습니다.
- 엑셀 파일, 구글 시트, BigQuery 쿼리 결과를 입력으로 받아 JSON 형태로 파싱하거나 직접적으로 JSON 형태의 입력을 받습니다.
- 쿼리(GET Parameter 또는 POST Body) 배열을 만들고, 반복문을 통해 각각의 쿼리를
fetch()메서드에 전달해 HTTP 요청에 사용합니다. - 컨텍스트 검증 또는 키-값 맵핑 과정을 각각 Query, Schema라는 배열 형태로 구조화하고, Info라는 딕셔너리형 클래스로 관리합니다.
- 여러 개의 Spider 객체를 동기적으로 연속해서 또는 비동기적으로 동시에 실행하고 싶을 경우를 위해 Pipeline이라는 클래스를 추가합니다.
- 입력 : 외부 입력 [
read_excel(),read_gspread(),read_gbq()]- 문제) 마케팅 담당자가 직접 검색 키워드와 같은 쿼리 배열을 입력하고 싶을 수 있습니다.
- 해결) 사용자 입력을 외부 데이터 소스를 읽어서 동적으로 할당하도록 파싱하는 로직을 구현했습니다.
- 추출 : 세션 생성 [
@init_session()]- 동기적 요청의 경우 requests.Session 객체, 비동기적 요청의 경우 aiohttp.ClientSession 객체를 생성해 추출 과정에 사용합니다.
- 비동기적 요청 시에 asyncio.Semaphore 객체를 추가로 생성해 과도한 동시 요청을 제한합니다.
- 추출 : 세션 생성 후 로그인 [
@login_session()]- 다이어그램에는 표시하지 않았지만, 로그인이 필요한 경로가 있습니다. 이러한 경로에 HTTP 요청을 보낼 때 반복문마다 로그인을 수행하면,
- 비효율적이기도 하고 대상 페이지에서 비정상적인 행위로 보여질 수 있기 때문에 세선 생성 직후 로그인을 한 번 수행합니다.
- 이때,
@init_session()대신에@login_session()데코레이터를 사용하면 미리 정의된 로그인 과정을 처리할 수 있습니다.
- 추출 : 쿼리 배열 생성 [
_init_iterator()]- 쿼리 범위가 입력으로 주어질 경우 쿼리 배열로 바꿔줄 필요가 있습니다.
- 예시로, 시작일이 일주일 전, 종료일이 어제, 기간이 하루 단위일 경우, datetime.date 객체를 요소로 갖는 쿼리 배열이 만들어집니다.
- 이때, 키워드나 상품코드 같은 쿼리 배열이 추가로 주어질 경우, 여러 개의 쿼리 배열을 카티션 프로덕트 연산을 통해 모든 쿼리 조합을 배열로 만들어냅니다.
- 해당 쿼리 배열의 각 요소는 반복문을 거치며
fetch()로 전달됩니다.
- 변환 : 스키마 & 필드 [
map(),map_schema(),map_field()]- 키-값을 맵핑하는 과정은 파싱된 HTTP 응답 결과에서 특정 경로의 값을 꺼내 새로운 딕셔너리에 지정된 키에 넣는 과정입니다.
- 이때, 키 명칭과 값을 꺼내는 경로를 Field라는 딕셔너리형 클래스로 관리합니다. 또한, 성격이 비슷한 Field 객체들을 재사용하기 위해,
- Schema라는 리스트형 클래스로 Field 객체들을 묶어서 관리합니다. Field 클래스에는 한글 설명을 넣을 수 있어 주석의 역할을 겸합니다.
- 참고로,
parse()메서드는 HTTP 응답 객체를 키-값 맵핑하기 쉬운 형태로 가공하는 로직이 구현됩니다. 데이터 형태를 바꾸는 중간 과정일 뿐입니다.
- 그림 1-2의 ETL 다이어그램에서는 Spider라는 클래스 하나의 로직을 구현했지만, 실제론 Spider 클래스 안에 전부 구현되어 있지 않고, 유사한 메서드를 부모 클래스에서 구현해놓고 Spider 클래스가 상속받습니다.
- 전체적인 프레임워크의 상속 관계는 그림 1-3을 참고할 수 있습니다.
- 모든 클래스는 딕셔너리(CustomDict) 또는 리스트(CustomRecords)로부터 파생됩니다. 데이터 추출 과정에서 parse() 및 map() 메서드는 Spider 클래스 안에서 전부 구현하기에는 코드 길이에 문제도 있고 의도적으로 데이터 추출과 변환 과정을 구분하기 위해 Parser, Mapper라는 별도의 부모 클래스에서 구현하고 상속받도록 했습니다.
- 외부 데이터 입력과 적재 과정은 마찬가지로 Spider 클래스로부터 GoogleQueryReader, GoogleUploader 클래스로 구분했습니다.
- 동기식/비동기식 요청 여부에 따라 데이터 추출 로직은 크게 달라집니다. 동기식은 Spider, 비동기식은 AsyncSpider 클래스로 구분하고, 각각에 차별화된 데이터 추출 과정을 구현합니다. 데이터 변환 및 적재 과정을 별도의 클래스로 구분한 덕분에 동일한 로직이 중복 구현될 걱정은 없습니다.
- 마찬가지로, 로그인 여부에 따라 비로그인은 Spider, 로그인은 EncryptedSpider 클래스로 구분했습니다. 그리고, 두 클래스에서 공통되는 컨텍스트 검증 등의 로직은 RequestSession이라는 부모 클래스로 분리하고 상속받게 했습니다.
- 로그인 과정은 LoginSpider 클래스의 login() 메서드에서 구현하고, EncryptedSpider 클래스는 LoginSpider 객체를 사용합니다.
- 그 외에 파라미터 검증과 주석의 역할을 겸하는 다양한 구조체들을 상속도의 하단에 표현했습니다.