존재해야 할 대상과 그것을 만드는 방법을 파이썬 함수로 적는다. @asset 표기가 붙은 함수를 Dagster가 모두 집어 올린다.
제6장에서 파이썬과 pandas·Arrow·Polars와의 매끄러운 통합을 살펴본 뒤라, DuckDB와 그 생태계가 데이터 파이프라인에 속한 여러 과제를 감당할 수 있고 따라서 그 안에서 쓰일 수 있음을 우리는 이미 안다. 강력한 SQL 엔진, 잘 통합된 도구, 그리고 클라우드 제공의 잠재력이 결합해 그것을 그저 또 하나의 데이터베이스 시스템 이상으로 만든다.
이 장은 먼저 한 걸음 물러나 데이터 파이프라인의 뜻과 중요성을 논한다. 그다음 튼튼한 파이프라인을 지을 때 도움이 된다고 저자들이 판단한 도구 몇 개를 살펴본다. 수집, 변환, 조율 셋을 아우르는 도구들이다.
NOTE
외부의 살아 있는 원천과 갱신되어온 저장소에서 데이터를 적재하므로, 이 장의 예제는 여러분이 코드를 돌릴 때와 다른 레코드 수나 개수를 보여줄 수 있다. 뒤에 나올 숫자들은 저자들이 집필하던 시점의 눈금이다.
DuckDB는 파이프라인에서 저장 시스템으로 쓰일 수도 있지만, 단순하면서도 강력한 실행 모델 덕에 대개 변환과 처리 부분에 앉는다.
8.1절, 이 장의 자리 정하기
이 장에 실린 목록 21건 — 언어가 셋이다. Python, YAML, 그리고 SQL
- 8.1새 dlt 파이프라인에 필요한 라이브러리 가져오기
- 8.2dlt 파이프라인 정의하기
- 8.3dlt 소스 정의하기
- 8.4소스에서 관심 있는 데이터셋 고르기
- 8.5dlt 파이프라인 실행하기
- 8.6다른 데이터셋으로 파이프라인 실행하기
- 8.7모든 출력에 DuckDB를 쓰는 dbt 프로필 정의
- 8.8웹 위치에서 데이터를 가져오는 dbt 소스 정의
- 8.9여러 질의로 데이터를 변환하는 dbt 모델 정의
- 8.10데이터셋의 여러 성질을 주장하는 dbt 스키마
- 8.11스키마에 테스트 더하기
- 8.12CSV 파일 목록을 파이프라인의 소스로 계산하기
- 8.13스키마에 세부 더하기
- 8.14새 파일들의 별스러움에 맞춰 변환 고치기
- 8.15Dagster 파이프라인에서 처리할 첫 애셋 정의
- 8.16
__init__.py에 Dagster 작업 정의 - 8.17외부 원천의 데이터를 DuckDB로 변환하는 애셋
- 8.18기존 데이터를 변환하는 애셋
- 8.19새 값을 계산해 새 데이터를 파생하는 애셋
- 8.20Dagster 작업에 새 정의 더하기
- 8.21DuckDB 저장 위치를 MotherDuck으로 바꾸기
8.1
데이터 파이프라인과 DuckDB의 역할Data pipelines and the role of DuckDB
데이터 파이프라인은 통상 여러 원천에서 데이터를 가져와 데이터 저장소로 수집하기 위해 세워진다. 저장소는 데이터베이스일 수도, 클라우드의 평면 파일로 저장된 데이터 레이크일 수도, 데이터 웨어하우스일 수도 있다. 무엇을 저장하기 전에 데이터는 통상 여러 방식으로 처리되고 변환된다. 데이터셋을 서로 조인하고, 걸러내고, 집계하고, 또는 마스킹하는 것이며, 목표는 온전한 통합과 표준화다.
앞의 세 가지는 제3·4장에서 넉넉히 다뤘지만 마스킹은 새로 나온 말이다. 마스킹은 기밀 데이터나 규제 대상 데이터를 익명화하거나 왜곡하는 일이다. 파이프라인이 다루는 것이 늘 공개 데이터만은 아니라는 사실이 이 한 낱말에 담겨 있다.
그리고 이 절의 방향을 정하는 문장이 나온다. 데이터를 저장하는 것만으로는 충분하지 않다. 파이프라인은 값을 만드는 일이다. 데이터 파이프라인의 어떤 쓰임도 결국은 제품—대시보드, API, 기계학습 모델 따위—을 만드는 일이다. 〈도판 8.1〉이 그 흐름과 방향을 보여준다.
배치와 스트리밍, 그리고 ETL과 ELT
다뤄야 할 데이터 파이프라인은 통상 두 가지 주된 유형이다. 배치 처리와 스트리밍 데이터이며, 이 장은 배치 처리를 다루는 파이프라인을 본다. 배치 처리는 통상 명령의 연쇄로 이루어진 작업 흐름을 만들며, 한 명령의 출력이 다음 명령의 입력이 된다. 마지막 변환이 끝나고 데이터가 원하는 저장소에 저장되면 처리가 완료된다.
언제 배치가 알맞은가. 모든 변화를 즉시 분석하거나 즉각적인 변화에 반응할 필요가 없을 때다. 판단의 기준을 기술이 아니라 요구에 두는 이 정의가 정확하다.
NOTE — ETL과 ELT
추출·변환·적재(ETL)의 흐름은 데이터 파이프라인의 하위 범주다. 모든 파이프라인이 그 정확한 순서를 따르는 것은 아니다. 대부분의 경우 추출이 첫 단계이기는 하지만, 데이터를 원하는 저장소에 먼저 적재하고 나서 변환할 수도 있다. 그 순서를 ELT라 부른다.
그리고 이 구분이 언제 중요해지는지를 짚는다. MotherDuck 같은 클라우드 서비스를 써서 자원을 최대한 잘 활용하고 싶을 때다. 때로는 로컬 자원으로 데이터를 변환하는 편이 낫고, 때로는 이미 클라우드에 저장된 데이터를 변환하는 편이 효과적이다. 제7장의 MD_RUN 논의가 여기서 되돌아온다.
공유하는 언어가 둘 있다
그래서 DuckDB의 역할은 무엇인가. 저장 시스템으로 쓰일 수도 있지만 대개 변환과 처리 부분에 앉는다. 이유는 실행 모델의 단순함과 힘이다. 대용량 데이터셋을 다룰 수 있는 단일 바이너리, 온갖 원천과 저장 형식을 입력으로 받는 능력, 그리고 데이터를 여러 방식으로 변환하는 완전한 SQL 엔진이다.
여기서 이 장 전체를 관통하는 통찰이 나온다.
- 첫째 공유 언어는 SQL이다 — 폭넓은 SQL 지원이 dbt 같은 관련 처리 도구와 통합할 첫 공유 언어를 제공한다.
- 둘째 공유 언어는 Parquet이다 — 특히 데이터 레이크에 저장할 때, 출력 형식으로 Parquet을 쓰는 일이 잦다.
도구가 서로 붙는 자리에는 언제나 공유 언어가 있어야 한다. 이 장에서 dlt와 dbt와 Dagster가 DuckDB에 붙는 방식이 모두 그 둘 가운데 하나로 설명된다. 제5장에서 Parquet의 스키마 내장과 푸시다운을 배워둔 것이 여기서 값을 한다.
8.2
dlt로 데이터 수집하기Data ingestion with dlt
dlt(data load tool)는 여러 가지, 그리고 흔히 지저분한 데이터 원천에서 다양한 목적지로 데이터를 적재하게 해주는 오픈소스 파이썬 라이브러리다. 그런데 왜 직접 파이썬 스크립트를 쓰지 않고 dlt를 쓰는가. 저자들의 답이 명확하다.
- 스키마 추론 — dlt의 주 입구인
pipeline함수가 원천 데이터에서 스키마를 추론하고 그 데이터를 목적지에 적재하면서 거기에 알맞은 스키마를 만든다. - 원천에 무관한 후속 처리 — JSON 데이터, 데이터프레임, 또는 생성기 함수 같은 다른 순회 가능한 객체와 함께 이 파이프라인을 쓸 수 있으며, 뒤따르는 처리를 하나도 바꾸지 않아도 된다.
- 버전 관리 — 엔진이 버전 관리까지 맡아준다. 그래서 데이터 팀은 데이터를 쓰고 값을 만드는 일에 집중하면서, 변경이 있을 때 적시에 알림을 받아 효과적인 거버넌스를 확보할 수 있다.
dlt는 SQL 데이터베이스, GitHub, 그 밖의 흥미로운 API를 포함한 미리 정의된 원천과 목적지 한 벌을 제공한다. 기본으로 지원되는 목적지 가운데 하나가 DuckDB다. 맞춤 원천과 목적지도 정의할 수 있지만 이 책에서는 다루지 않는다.
$ pip install dlt $ pip install duckdb # 제6장을 건너뛰었다면 이것도 필요하다 # dlt init은 소스와 목적지 두 인수를 받는다. $ dlt init chess duckdb Looking up the init scripts in https://github.com/dlt-hub/verified-sources.git... Cloning and configuring a verified source chess (A source loading player profiles and games from chess.com api) Do you want to proceed? [Y/n]: Verified source chess was added to your project! * See the usage examples and code snippets to copy from chess_pipeline.py * Add credentials for duckdb and other secrets in ./.dlt/secrets.toml * requirements.txt was created. Install it with: pip install -r requirements.txt
이 명령은 실행 가능한 스크립트를 포함해 파일 몇 개를 로컬에 만든다. 그래서 진행 여부를 확인받는다.
끝나면
chess 디렉터리에 보조 함수들과, 작동하는 표본 파이프라인인 chess_pipeline.py가 놓인다.
원하는 소스와 목적지가 모두 공식 지원되므로 명령 한 줄로 필요한 파일과 정의가 다 만들어진다.
# 목록 8.1 — 필요한 라이브러리를 가져온다 import dlt from chess import source # 목록 8.2 — 파이프라인을 정의한다 pipeline = dlt.pipeline( pipeline_name="chess_pipeline", # 이 이름이 만들어질 DuckDB 파일 이름에 반영된다 destination="duckdb", dataset_name="main" # DuckDB의 스키마 이름이 된다. main은 기본 스키마다 ) # 목록 8.3 — 가장 인기 있는 기사 넷과 2022년 11월 대국을 소스로 만든다 data = source( players=[ "magnuscarlsen", "vincentkeymer", "dommarajugukesh", "rpragchess" ], start_month="2022/11", # 이 인수들은 chess.com API에 고유한 것이다 end_month="2022/11", ) # 목록 8.4 — 소스가 담은 여러 리소스 가운데 프로필만 고른다 players_profiles = data.with_resources("players_profiles") # 목록 8.5 — 여기까지는 정의뿐이었다. 이제 실행한다 info = pipeline.run(players_profiles) print(info) Pipeline chess_pipeline completed in 0.62 seconds 1 load package(s) were loaded to destination duckdb and into dataset main The duckdb destination used duckdb:////path/to/code/ch08/dlt_example/chess_pipeline.duckdb location to store data Load package 1696519035.883884 is LOADED and contains no failed jobs
DuckDB 데이터베이스는 <파이프라인-이름>.duckdb라는 파일로 기록된다. 여기서는
chess_pipeline.duckdb다.
이 소스는 프로필, 대국, 접속 상태 등 선수와 관련된 여러 리소스를 담고 있어, 그중 필요한 것만 골라 쓴다.
# 파이썬 셸을 나가지 말고, 두 번째 터미널을 열어 데이터베이스를 적재한다. $ duckdb chess_pipeline.duckdb D SHOW TABLES;
| name |
|---|
| varchar |
| _dlt_loads |
| _dlt_pipeline_state |
| _dlt_version |
| players_profiles |
dlt가 자기 메타데이터를 저장할 테이블 한 무리를 함께 만들어두었다.
맨 아래에 우리가 지정한 선수들의 프로필을 담고 있을
players_profiles가 보인다.
밑줄로 시작하는 세 테이블이 뒤에서 결정적인 구실을 한다.
D .mode line D FROM players_profiles LIMIT 1; avatar = https://images.chesscomfiles.com/uploads/v1/user/… player_id = 138850604 aid = https://api.chess.com/pub/player/dommarajugukesh url = https://www.chess.com/member/DommarajuGukesh name = Gukesh Dommaraju username = dommarajugukesh followers = 3 country = https://api.chess.com/pub/country/IN location = Chennai last_online = 2022-07-16 19:18:02+01 joined = 2021-05-05 10:27:46+01 status = basic is_streamer = false verified = false league = Wood _dlt_load_id = 1696519035.883884 _dlt_id = kldRaeRA40OGBA title =
구케시 도마라주의 프로필이니 파이프라인은 여기까지 잘 작동한다.
_dlt_load_id와 _dlt_id 두 컬럼이 데이터 행마다 붙어 있다는 점을 눈여겨볼 만하다.
어느 적재 묶음에서 온 행인지를 데이터 자체가 지니고 있어야 계보(lineage)를 되짚을 수 있다.
# 앞에서처럼 리소스를 변수에 할당하지 않고 run 메서드에 그냥 넘긴다는 점을 눈여겨볼 것. info = pipeline.run(data.with_resources("players_profiles", "players_games")) print(info) Getting archive from https://api.chess.com/pub/player/magnuscarlsen/games/2022/11 Getting archive from https://api.chess.com/pub/player/vincentkeymer/games/2022/11 Getting archive from https://api.chess.com/pub/player/rpragchess/games/2022/11 Pipeline chess_pipeline completed in 1.89 seconds 1 load package(s) were loaded to destination duckdb and into dataset main Load package 1696519484.186974 is LOADED and contains no failed jobs
| count_star() |
|---|
| int64 |
| 589 |
duckdb chess_pipeline.duckdb 'SELECT count(*) FROM players_games'로 확인한 결과다.
대국이 잘 수집되었고, 같은 저장소를 그대로 썼다.
요청한 선수는 넷인데 출력에 보이는 아카이브는 셋이다.
dommarajugukesh의 아카이브를 가져오는 줄이 없다.
원서 본문은 "관심 있던 선수들의 아카이브 여러 개를 적재했음을 알려준다"고만 적었으니,
출력이 잘렸거나 그 선수의 해당 월 대국이 없었을 것이다. 어느 쪽인지는 본문만으로 알 수 없다.
그리고 이 절에서 가장 인상적인 대목. 파이프라인을 다시 돌리면 어떻게 되는가.
dlt는 요청한 것 가운데 사용 가능한 모든 것을 이미 수집했음을 알아내고 새로 가져오지 않는다.
저자들의 평가가 솔직하다. "이것은 실로 인상적이다. 배치 수집을 멈춘 자리에서 이어받는 일은 쉬운 과제가 아니다."
그것을 통제하는 데 필요한 정보가 앞서 본 세 개의 밑줄 테이블에 들어 있다.
$ dlt pipeline chess_pipeline info Found pipeline chess_pipeline in /Home/.dlt/pipelines Synchronized state: _state_version: 2 _state_engine_version: 2 schema_names: ['chess'] pipeline_name: chess_pipeline destination: dlt.destinations.duckdb default_schema_name: chess staging: None dataset_name: main Local state: first_run: False _last_extracted_at: 2023-11-04T19:16:35.873231+00:00 Resources in schema: chess players_profiles with 1 table(s) and 0 resource state slot(s) players_games with 1 table(s) and 1 resource state slot(s) Working dir content: Has 3 completed load packages with following load ids: 1699125395.876516 1699125399.292224 1699125402.854308 Pipeline has last run trace. Use 'dlt pipeline chess_pipeline trace' to inspect
info 명령은 파이프라인 정의와 저장소 안의 메타 테이블을 모두 살펴 파이프라인 자신의 상태에 대한 시야를 되돌려준다.
완료된 적재 묶음이 셋이니 파이프라인을 세 번 돌렸다는 뜻이다.
두 이름이 나란히 나오는 것이 헷갈릴 만하다. dataset_name은 main이고 default_schema_name은 chess다.
전자는 DuckDB 쪽 스키마 이름이고 후자는 dlt 자신이 관리하는 스키마의 이름이다.
같은 낱말이 두 세계에서 다른 것을 가리키는 자리이니 구분해두어야 한다.
players_games에는 리소스 상태 슬롯이 하나 있고 players_profiles에는 없다.
증분 수집을 위해 "어디까지 읽었는지"를 기억해야 하는 리소스에만 그 슬롯이 생긴다.
앞에서 본 "이어받기"의 실체가 이 한 줄에 드러난다.
8.3
dbt로 데이터 변환하고 모델링하기Data transformation and modeling with dbt
날것 그대로의 데이터는 참된 잠재력을 끌어내기 위해 흔히 모양 잡기, 정리, 모델링을 요구한다. dbt(data build tool)는 데이터 파이프라인의 생성과 관리를 지원하도록 설계된 SQL 중심의 변환 도구다. 그리고 그 성격을 규정하는 문장이 이어진다. dbt는 소프트웨어 공학의 원칙을 강조해, 데이터 팀이 모듈성·이식성·문서화를 확보할 수 있게 한다. dbt 안에 CI/CD를 통합하면 데이터 변환의 일관되고 신뢰할 수 있는 배포가 쉬워진다.
DuckDB와는 dbt-duckdb 라이브러리가 다리를 놓는다. 두 도구의 강점을 결합해, dbt로 구동되는 데이터 파이프라인에서 변환을 적용하는 데 DuckDB를 쓸 수 있게 해준다. 그리고 dbt와 DuckDB가 함께 말하는 공유 언어는 물론 SQL이다.
이 절에서 지을 것은 명확하다. GitHub에 저장된 CSV 파일 몇 개를 가져와 정리와 변환을 적용하고, 정제된 데이터를 담은 Parquet 파일을 출력한다. 따라서 DuckDB는 저장소로 쓰이지 않고 변환의 수단으로만 쓰인다. 8.1절이 말한 바로 그 자리다.
$ pip install dbt-duckdb dbt # 처음 돌리면 프로필을 설정하며 쓸 데이터베이스를 묻는다. 여기서 DuckDB를 고른다. $ dbt init dbt_transformations $ cd dbt_transformations $ tree . ├── README.md ├── analyses ├── dbt_project.yml ├── macros ├── models │ └── example │ ├── my_first_dbt_model.sql │ ├── my_second_dbt_model.sql │ └── schema.yml ├── seeds ├── snapshots └── tests # 예제를 치우고 우리 디렉터리로 바꾼다. $ mv models/example models/atp $ rm models/atp/*.sql $ mkdir output
dbt는 프로젝트 단위로 생각한다. 주요 개념마다 폴더가 있다.
macros, models, seeds, snapshots, tests다.
이 표본 프로젝트에서는 models와 tests만 쓰고 나머지는 쓰지 않는다.
# 목록 8.7 — profiles.yml. 모든 출력에 DuckDB를 쓰는 프로필 dbt_transformations: target: dev outputs: dev: type: duckdb path: '/tmp/atp.db' # dbt가 쓰는 중간 DuckDB 저장소. 위치는 아무 데나 된다 schema: 'main' # 목록 8.8 — models/atp/sources.yml. 웹 위치에서 데이터를 가져오는 소스 version: 2 # 집필 시점 dbt 판본에서 작동하려면 2여야 한다 sources: - name: github # 소스 이름. 아무렇게나 지어도 된다 meta: external_location: 'https://raw.githubusercontent.com/JeffSackmann/ tennis_atp/master/atp_matches_2023.csv' tables: - name: matches_file # CSV 파일을 참조할 때 쓸 이름. 이것도 아무렇게나
소스 이름(github)과 테이블 이름(matches_file)의 조합으로 모델에서 이 소스를 참조하게 된다.
dbt 소스는 외부 데이터베이스나 웨어하우스, 그 밖의 어디에 있든 원자료를 참조하고 문서화하는 표준화된 방법을 제공하며,
원자료에 접근하는 방식의 일관성을 확보하면서 그 데이터셋에 딸린 메타데이터와 품질 검사도 함께 지정하게 해준다.
NOTE — YAML YAML Ain't Markup Language는 사람이 읽기 좋고 스크립팅 언어와 상호작용하도록 설계된
데이터 직렬화 언어다. 문법이 비교적 단순하고 들여쓰기로 계층을 표현한다.
그리고 그 목적에 탭이 아니라 오직 공백만 쓴다.
먼저 CSV 파일 하나만 처리한다. 2023년 파일이다.
기다리는 시간을 줄여, 스키마와 모델이 작동하는지에 대한 즉각적인 피드백을 얻으려는 것이라고 저자들이 밝혀두었다.
전체를 한 번에 붙잡지 않는 이 순서가 이 절의 실무 감각이다.
WITH noWinLoss AS ( SELECT COLUMNS(col -> NOT regexp_matches(col, 'w_.*') AND -- w_로 시작하는 컬럼을 걸러낸다 NOT regexp_matches(col, 'l_.*') -- l_로 시작하는 컬럼도 ) FROM -- 앞 절에서 정의한 소스 ) SELECT * REPLACE ( cast(strptime(tourney_date, '%Y%m%d') AS date) as tourney_date ) FROM noWinLoss -- tourney_date를 date 타입으로 강제한다
이 파일 하나에 제3장에서 배운 DuckDB 확장이 셋이나 들어 있다.
3.5.1절의 COLUMNS + 람다로 컬럼을 정규식으로 걸러내고,
SELECT * REPLACE로 한 컬럼만 갈아 끼우고, 제5장 5.4절의 strptime으로 8자리 숫자를 날짜로 되돌린다.
dbt는 SQL을 담는 그릇일 뿐이고, 실제 일은 DuckDB의 방언이 한다.
파이프라인이 돌 순서는 넷이다. ① dbt가 GitHub에서 CSV 데이터를 가져온다 ② 그 내용을 DuckDB에 넘긴다
③ SQL로 쓰인 변환을 실행한다 ④ 결과를 Parquet으로 저장한다.
$ dbt run 09:48:35 Found 1 model, 1 source, 0 exposures, 0 metrics, 351 macros, 0 groups, 0 semantic models 09:48:35 Concurrency: 1 threads (target='dev') 09:48:35 1 of 1 START sql external model main.matches_2023 ......... [RUN] 09:48:37 1 of 1 OK created sql external model main.matches_2023 .... [OK in 1.16s] 09:48:37 Finished running 1 external model in 0 hours 0 minutes and 1.20 seconds 09:48:37 Completed successfully 09:48:37 Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1 $ du -h output/* 120K output/matches.parquet
실행 출력이 만든 모델의 이름을 main.matches_2023이라고 부른다.
그런데 모델 파일은
models/atp/matches.sql이고, 곧 나올 목록 8.10의 스키마도
- name: matches라고 선언한다. 모델 이름은 파일 이름에서 오므로 matches여야 한다.
집필 도중 파일을 matches_2023.sql에서 matches.sql로 고치면서
출력만 예전 것이 남은 것으로 보인다.
| count_star() |
|---|
| int64 |
| 2986 |
SELECT count(*) FROM 'output/matches.parquet';의 결과다.
집필 시점에는 2,986건이지만 경기가 계속 추가되므로 이 숫자는 더 클 수 있다.
이 2,986이라는 숫자를 기억해두어야 한다. 8.3.5절에서 결정적인 구실을 한다.
D .mode line D FROM 'output/matches.parquet' LIMIT 1; tourney_id = 2023-9900 tourney_name = United Cup surface = Hard tourney_level = A tourney_date = 2023-01-02 match_num = 300 winner_name = Taylor Fritz winner_hand = R winner_ht = 193 winner_ioc = USA winner_age = 25.1 loser_name = Matteo Berrettini loser_hand = R loser_ht = 196 loser_ioc = ITA loser_age = 26.7 score = 7-6(4) 7-6(6) best_of = 3 round = F minutes = 135 winner_rank = 9 winner_rank_points = 3355 loser_rank = 16 loser_rank_points = 2375
w_와 l_로 시작하는 세밀한 경기 통계 필드가 사라졌고, tourney_date가 제대로 된 날짜가 되었다.
첫 dbt 파이프라인을 성공적으로 썼다. 그런데 Parquet 안의 데이터가 옳은지는 어떻게 아는가?
그 물음이 다음 절을 부른다.
tourney_level = A를 표시해두었다. United Cup은 국가 대항 팀 대회다.
그런데 8.4.4절의 목록 8.19는 코드 A를 "Masters 1000s"로 짝지어놓는다.
이 어긋남을 뒤에서 따로 다룬다.
8.3.4
변환과 파이프라인 테스트하기Testing transformations and pipelines
dbt 테스트는 데이터 품질과 일관성을 보장하기 위해 데이터 모델에 적용하는 주장(assertion)이다. 이 테스트를 정의함으로써 변환을 검증하고, NULL 값, 중복, 참조 정합성 위반 같은 문제를 잡아낼 수 있다.
테스트를 정의할 수 있는 한 자리가 models/atp/schema.yml이다. 모델 파일 옆에 놓인다. 여기서는 컬럼 몇 개에 대해서만 테스트를 만들지만, 운영 파이프라인이라면 변환이 기대대로 작동했는지 확인하기 위해 모든 필드에 테스트를 만들고 싶을 것이다.
version: 2 models: - name: matches description: "ATP tennis matches schema" columns: - name: tourney_id description: "The ID of the tournament." tests: - not_null - name: winner_id tests: [ not_null ] - name: loser_id tests: [ not_null ] - name: surface description: "The surface of the court." tests: - not_null - accepted_values: values: ['Grass', 'Hard', 'Clay'] # 목록 8.11 — dbt_expectations 패키지를 더해 더 촘촘한 주장을 얹는다 - name: tourney_date description: "Verify that the tournament started in 2023" tests: - dbt_expectations.expect_column_values_to_be_of_type: column_type: date - dbt_expectations.expect_column_min_to_be_between: min_value: "CAST('2023-01-01' AS DATE)" max_value: "CAST('2023-12-31' AS DATE)"
검증할 네 가지 주장이다. tourney_id·winner_id·loser_id가 NULL이 아니고, surface가 NULL이 아니면서
Grass·Hard·Clay 가운데 하나만 담는다.
저자들의 조언이 실용적이다. 모든 필드에 테스트를 만들 필요는 없다. 쓰임에 필요한 것만 만든다.
데이터가 지저분할 수 있는 필드나, 지저분해도 상관없는 필드에는 테스트를 만들지 않는다.
테스트가 많을수록 좋다는 통념을 정면으로 거스르는 말인데, 파이프라인을 실제로 운영해본 사람의 말이다.
내장 주장으로 부족할 때는 패키지를 쓴다. 최상위에
packages.yml을 만들어
calogica/dbt_expectations 0.10.1을 적고 dbt deps를 돌리면
dbt_expectations와 그 의존인 dbt_date까지 설치된다.
$ dbt test 10:57:39 Found 1 model, 5 tests, 1 source, 0 exposures, 0 metrics, 351 macros 10:57:39 1 of 5 PASS accepted_values_matches_surface__Grass__Hard__Clay [0.14s] 10:57:39 2 of 5 PASS not_null_matches_loser_id [0.12s] 10:57:39 3 of 5 PASS not_null_matches_surface [0.12s] 10:57:39 4 of 5 PASS not_null_matches_tourney_id [0.12s] 10:57:39 5 of 5 PASS not_null_matches_winner_id [0.12s] 10:57:39 Finished running 5 tests in 0 hours 0 minutes and 0.66 seconds
테스트 이름이 자동으로 만들어진다는 점을 눈여겨볼 만하다.
accepted_values_matches_surface__Grass__Hard__Clay처럼
주장 종류 + 모델 + 컬럼 + 값이 이름에 다 들어간다.
실패했을 때 어느 주장이 깨졌는지 이름만 보고 알 수 있게 하려는 설계다. 다음 절에서 그 값을 한다.
8.3.5
모든 CSV 파일 변환하기Transforming all CSV files
지금까지는 2023년 경기만 다뤘지만, 처리해야 할 CSV 파일이 1968년까지 거슬러 올라간다. 소스를 먼저 바꾼다. external_location이 1968년부터 2023년까지 연도를 순회하며 모든 CSV 파일 URL의 목록을 만들고 read_csv_auto로 그것들을 읽는 함수를 돌려주도록 바꾸는 것이다. 저자들의 말대로 이것이 이 파이프라인에서 DuckDB가 저장이 아니라 처리 도구로 쓰이는 전형적인 예다.
version: 2 sources: - name: github meta: external_location: > (FROM read_csv_auto( list_transform( range(1968, 2023), # 1968년부터의 연도 목록을 만든다 y -> 'https://raw.githubusercontent.com/JeffSackmann/ tennis_atp/master/atp_matches_' || y || '.csv' ), # 각 연도를 URL로 바꾼다 types={'winner_seed': 'VARCHAR', 'loser_seed': 'VARCHAR'} )) formatter: oldstyle # {} 문자를 쓸 수 있게 하려면 oldstyle 포매터가 필요하다 tables: - name: matches_file
YAML의
>는 여러 줄을 한 줄로 접어 넣는 블록 표기다. 그 안에 DuckDB SQL 조각이 그대로 들어간다.
list_transform과 람다는 제4장에서, range는 4.9절에서 배운 것이다.
YAML 설정 파일 안에서 SQL이 돌고, 그 SQL이 다시 URL을 계산한다. 층이 셋이나 겹친 자리다.
range(1968, 2023)은 2023년을 포함하지 않는다.
제4장 4.9절이 못 박아둔 규칙이다. range의 상한은 배타적이고 generate_series의 상한은 포함적이다.
그러므로 이 목록은 1968년부터 2022년까지 55개 연도이며, 절의 제목이 "모든 CSV 파일"인데도
가장 최근 연도가 빠진다. 뒤에 이어지는 산수가 그것을 증명한다.
$ dbt run D SELECT count(*) FROM 'output/matches.parquet'; -- → 188,934 $ dbt test 12:58:05 Finished running 7 tests in 0 hours 0 minutes and 1.49 seconds 12:58:05 Completed with 3 errors and 0 warnings: 12:58:05 Failure in test accepted_values_matches_surface__Grass__Hard__Clay 12:58:05 Got 1 result, configured to fail if != 0 12:58:05 Failure in test dbt_expectations_expect_column_min_to_be_between_ matches_tourney_date__CAST_2023_12_31_AS_DATE___CAST_2023_01_01_AS_DATE_ 12:58:05 Got 1 result, configured to fail if != 0 12:58:05 Failure in test not_null_matches_surface 12:58:05 Got 2937 results, configured to fail if != 0 12:58:05 Done. PASS=4 WARN=0 ERROR=3 SKIP=0 TOTAL=7
깨진 셋이 각각 무엇을 뜻하는지가 이름에 다 있다.
surface에 NULL이 있다. Grass·Hard·Clay가 아닌 surface가 있다. 그리고 2023년이 아닌 날짜의 경기가 있다.
마지막 것은 예측 가능했다. 이제 2023년만이 아니라 모든 연도의 데이터를 들이고 있으니까.
D FROM 'output/matches.parquet' SELECT surface, count(*) GROUP BY ALL;
| surface | count_star() |
|---|---|
| varchar | int64 |
| Clay | 67537 |
| Carpet | 20900 |
| Hard | 74814 |
| Grass | 22746 |
| 2937 |
Carpet은 타당한 값처럼 보이지만, surface가 없는 행도 3,000건에 가깝다.
그래서
schema.yml은 Carpet을 허용하도록 고치고, matches.sql 모델은
surface가 NULL인 경기를 걸러내도록 고친다.
다섯 값을 더해보면 67,537 + 20,900 + 74,814 + 22,746 + 2,937 = 188,934다.
위에서 센 총계와 정확히 맞는다. 원서의 두 출력이 서로 아귀가 맞는다는 뜻이니,
이 지점까지의 숫자는 믿고 다음 검산으로 넘어갈 수 있다.
# 목록 8.13 — 스키마에 세부를 더한다 - name: surface tests: - not_null - accepted_values: values: ['Grass', 'Hard', 'Clay', 'Carpet'] # Carpet도 허용한다 - name: tourney_date tests: - dbt_expectations.expect_column_min_to_be_between: min_value: "CAST('1967-12-01' AS DATE)" max_value: "CAST('2023-12-31' AS DATE)"
하한이 1968년 1월 1일이 아니라 1967년 12월 1일인 점이 섬세하다.
1968년 첫 대회가 1967년 12월에 시작했을 수 있기 때문이다.
연도로 파일을 고르는 것과 날짜로 값을 검증하는 것은 다른 일이다.
-- 목록 8.14 — 새 파일들의 별스러움에 맞춰 변환을 고친다 SELECT * REPLACE ( cast(strptime(tourney_date, '%Y%m%d') AS date) as tourney_date ) FROM noWinLoss WHERE surface IS NOT NULL -- surface가 없는 행을 걸러낸다
모델이 바뀌었으니 dbt test 전에 dbt run을 다시 돌려야 모든 테스트가 통과한다.
GitHub의 CSV 파일에서 시작해 단일 Parquet 파일로 성공적으로 변환하고, 그 과정에서 데이터를 정제했다.
NOTE 다음 걸음은 개발용 파이프라인과 나란히 운영용 파이프라인을 세우는 일이다.
상당히 비슷하겠지만, 아마 Parquet 파일을 로컬 파일시스템이 아니라 S3 버킷에 쓰게 될 것이다.
저자들은 이것을 독자의 연습으로 남겨둔다.
살펴본 김에 — 산수가 증명하는 것
8.4절의 Dagster 파이프라인은 같은 데이터셋을 파이썬의 range(1968, 2024)로 읽어 191,920행을 얻는다. 이쪽 dbt 파이프라인은 DuckDB의 range(1968, 2023)로 읽어 188,934행을 얻었다. 차이는 2,986행이다.
그리고 8.3.3절이 2023년 파일 하나만 읽었을 때 센 경기 수가 정확히 2,986건이었다. 두 숫자가 한 자리도 어긋나지 않고 일치한다.
데이터가 계속 늘어나는 원천이므로 이 일치가 우연일 가능성을 완전히 배제할 수는 없다. 그러나 range의 상한이 배타적이라는 제4장의 규칙과 맞물려 보면 결론은 하나다. dbt 파이프라인은 "모든 CSV 파일"이라는 절 제목에도 불구하고 2023년을 빠뜨렸다. 고치려면 range(1968, 2024)로 쓰거나 generate_series(1968, 2023)로 바꾸면 된다.
8.4
Dagster로 데이터 파이프라인 조율하기Orchestrating data pipelines with Dagster
여기까지 배운 것은 외부 원천에서 DuckDB로 데이터를 적재하거나 DuckDB로 형식 사이를 변환하는 도구들이다. 중요한 과제이지만 퍼즐의 한 조각이 아직 빠져 있다. 그 변환이나 수집 코드를 무엇이 촉발하고 조율하는가?
조율 도구가 없는 세상이라면 손수 스케줄링과 실행 코드를 써야 한다. dbt 명령을 돌릴 cron 작업을 설정하고, dbt 태스크의 순서와 의존을 처리할 맞춤 스크립트를 써야 한다. 다행히 Airflow, Luigi, Kestra, Prefect, 그리고 이 장에서 쓸 Dagster 같은 도구가 존재한다.
Dagster의 핵심 개념 하나를 정확히 옮겨둘 만하다. 애셋(software-defined asset)이다. 테이블, 파일, 기계학습 모델처럼 영속 저장소에 있는 객체를 가리키며, 소프트웨어로 정의된 애셋은 존재해야 하는 애셋과 그것을 생산하고 갱신하는 방법에 대한, 코드로 된 서술이다. 애셋들이 작업(job)의 부분을 이루고, 그 작업은 스케줄링될 수 있다.
애셋 asset
작업 job
애셋들을 묶는다. define_asset_job과 AssetSelection.all()로 전부를 고를 수 있다.
스케줄 schedule
ScheduleDefinition에 cron 문법으로 주기를 준다. "0 * * * *"는 매시 정각이다.
왜 파이썬 함수로 서술하는가. 저자들의 논거가 분명하다. 애셋을 파이썬 함수로 서술하면 의존과 상호작용을 검증 가능한 코드로 서술할 수 있으며, 이는 설정 기반 도구에 비해—특히 개발자에게—큰 이점이다. 그리고 Dagster는 데이터 계보(lineage)와 데이터 출처(provenance)를 특히 잘 지원하는데, 그것이 감사·디버깅·추적을 위해 데이터 파이프라인의 중요한 측면이기 때문이다.
이번에는 Parquet 파일을 만드는 대신 데이터를 DuckDB에 적재한다. 경기 외에 선수 프로필도 들이고, pandas 데이터프레임에서 정적 데이터를 적재하는 법도 본다.
SIDEBAR — dagster-webserver 설치 때 만날 수 있는 오류
집필 시점에 dagster-webserver 패키지를 설치하려 할 때 다음 오류가 나면서 Dagster 서버가 내려갈 수 있다.
ImportError: cannot import name 'appengine' from
'requests.packages.urllib3.contrib'
....
raise Exception(
Exception: dagster-webserver process shut down unexpectedly
with return code 1
이 오류를 만나면 의존 urllib3과 requests-toolbelt을 다음 판본으로 못 박아야 한다.
pip install urllib3==1.26.15 requests-toolbelt==0.10.1
$ pip install dagster dagster-duckdb dagster-webserver # atp/__init__.py — 조율 코드가 들어간다 # atp/assets.py — 애셋 정의 코드가 들어간다 from dagster_duckdb import DuckDBResource from dagster import asset @asset def atp_matches_dataset(duckdb_resource: DuckDBResource) -> None: base = "https://raw.githubusercontent.com/JeffSackmann/tennis_atp/master" csv_files = [ f"{base}/atp_matches_{year}.csv" # CSV 파일 목록을 만든다 for year in range(1968,2024) ] create_query = """ CREATE OR REPLACE TABLE matches AS SELECT * REPLACE( cast(strptime(tourney_date, '%Y%m%d') AS date) as tourney_date ) FROM read_csv_auto($1, types={ 'winner_seed': 'VARCHAR', 'loser_seed': 'VARCHAR', 'tourney_date': 'STRING' }) """ with duckdb_resource.get_connection() as conn: conn.execute(create_query, [csv_files]) # $1이 이 목록을 받는다
$1은 execute 함수에 넘긴 첫 매개변수를 가리킨다. 여기서는 CSV 파일 목록이다.
파이썬이 URL 목록을 만들고, SQL이 그 목록 전체를 한 번에 읽는다.
제5장에서 배운 여러 파일을 한꺼번에 읽는 능력이 여기서 쓰인다.
매개변수 이름이 duckdb_resource다. 그런데 목록 8.16의
Definitions는 리소스 키를
"duckdb"로 등록하고, 이어지는 목록 8.17·8.18·8.19는 모두 duckdb라는 이름을 쓴다.
Dagster는 매개변수 이름으로 리소스를 찾아 주입하므로, 이 애셋만 이름이 어긋나 있으면
리소스를 찾지 못한다. 목록 8.15의 duckdb_resource를 duckdb로 고쳐야 나머지와 아귀가 맞는다.
from dagster_duckdb import DuckDBResource from dagster import ( AssetSelection, ScheduleDefinition, Definitions, define_asset_job, load_assets_from_modules, ) from . import assets atp_job = define_asset_job("atp_job", selection=AssetSelection.all()) atp_schedule = ScheduleDefinition( # 매시 한 번 도는 스케줄이다 job=atp_job, cron_schedule="0 * * * *", # cron 문법은 crontab.guru 참고 ) all_assets = load_assets_from_modules([assets]) # @asset 붙은 함수를 다 집어 온다 defs = Definitions( # 애셋·작업·리소스·스케줄을 한자리에 묶는다 assets=all_assets, jobs=[atp_job], resources={"duckdb": DuckDBResource( database="atp.duckdb", # Dagster를 돌린 위치를 기준으로 한 상대 경로 )}, schedules=[atp_schedule], )
이 파일이 쓸 수 있는 라이브러리를 통제하고, 작업을 언제 돌릴지 결정하고,
필요하면 환경에서 정보를 가져온다.
load_assets_from_modules가 표기만 붙여두면 알아서 모아 오는 방식이므로,
새 애셋을 더할 때 이 파일을 고칠 필요가 없다.
$ dagster dev -m atp # 브라우저에서 http://localhost:3000 으로 간다. # 화면 위의 Materialize를 누르면 몇 초 뒤 애셋이 실체화된다. $ duckdb atp.duckdb 'SELECT count(*) FROM matches'
| count_star() |
|---|
| int64 |
| 191920 |
Dagster는 파이프라인을 실체화한 뒤 데이터베이스를 열어두거나 잠그지 않으므로,
DuckDB 데이터베이스에 접근하기 전에 어떤 프로세스도 멈출 필요가 없다.
제7장에서 업로드 전에
.open md:로 잠금을 풀어야 했던 것과 대조된다.
NOTE 집필 시점에 191,920건이지만 이 코드를 시험해볼 때는 늘어나 있을 가능성이 크다.
0보다 크기만 하면 수집은 잘 된 것이다.
약 20만 건의 테니스 경기를 Dagster로 DuckDB에 성공적으로 적재했다.
〈도판 8.4〉·〈도판 8.5〉 · 화면 갈무리
원서의 이 두 도판은 Dagster UI의 화면 갈무리다. 앞의 것은 우리가 정의한 작업과 스케줄, 애셋을 보여주는 초기 화면이고, 뒤의 것은 데이터셋의 상태가 "실체화된 적 없음"에서 "실체화됨"으로 바뀌는 모습이다. 화면 갈무리는 이 문서에서 재현하지 않고 서술로 대신한다.
8.4.3
파이프라인의 의존 관계 관리하기Managing dependencies in a pipeline
지금까지는 경기 데이터만 들였다. 그런데 데이터베이스를 쓰는 애플리케이션은 경기에 참여한 선수들에 대한 다른 데이터도 필요할 수 있다. 그래서 선수 정보 적재를 파이프라인에 더하고, 선수 애셋과 이미 만든 경기 애셋 사이에 의존 관계를 만든다.
선수 파일에는 손볼 데가 있다. dob 컬럼의 값이 yyyymmdd 형식인데 마지막 네 자리가 0000인 행이 있다. 00은 유효한 월도 일도 아니므로 각각을 01로 기본 설정해서 strptime으로 날짜 타입으로 강제할 수 있게 한다. 이 애셋은 원천을 질의하기만 하는 것이 아니라 변환까지 함께 적용한다.
# 목록 8.17 — 외부 원천의 데이터를 DuckDB로 변환하는 애셋 @asset def atp_players_dataset(duckdb: DuckDBResource) -> None: base = "https://raw.githubusercontent.com/JeffSackmann/tennis_atp/master" csv_file = f"{base}/atp_players.csv" with duckdb.get_connection() as conn: conn.execute(""" CREATE OR REPLACE TABLE players AS SELECT * REPLACE( CASE WHEN dob IS NULL THEN NULL -- NULL 날짜는 그대로 둔다 WHEN SUBSTRING(CAST(dob AS VARCHAR), 5, 4) = '0000' THEN CAST(strptime( CONCAT(SUBSTRING(CAST(dob AS VARCHAR), 1, 4), '0101'), '%Y%m%d' ) AS date) -- 월·일을 0101로 채운다 ELSE CAST(strptime(dob, '%Y%m%d') AS date) -- 정상 형식은 그대로 파싱 END AS dob ) FROM read_csv_auto($1, types = { 'dob': 'STRING' }); """, [csv_file]) # 목록 8.18 — 기존 데이터를 변환하는 애셋. 의존을 선언한다. @asset(deps=[atp_players_dataset]) def atp_players_name_dataset(duckdb: DuckDBResource) -> None: concatenate_query = """ ALTER TABLE players ADD COLUMN name_full VARCHAR; UPDATE players SET name_full = name_first || ' ' || name_last """ with duckdb.get_connection() as conn: conn.execute(concatenate_query, [])
deps=[atp_players_dataset] 한 줄이 이 절의 주제다.
atp_players_dataset을 먼저 실체화하지 않고 atp_players_name_dataset을 실체화하려 하면 Dagster가 경고를 주고,
파이프라인을 돌릴 때는 그에 맞춰 애셋의 실행 순서를 정한다.
players 테이블이 없으면 컬럼을 더할 수 없으니 당연한 처사다.
atp_players_name_dataset은 외부 원천 데이터를 하나도 쓰지 않고 이미 저장소에 수집된 데이터를 변환한다.
이런 종류의 의존 애셋 연산은 데이터 리팩터링, 정제, 증강처럼
하나 이상의 애셋을 입력으로 삼아 작업을 실행하는 여러 쓰임에 쓸 수 있다.
이제 실행하면 무슨 일이 벌어지는가. 저자들이 남긴 실측 시간이 있으니, 선언한 의존이 실행 순서를 어떻게 바꾸는지 직접 확인할 수 있다. 아래에서 재생 버튼을 누르고, 의존을 선언하지 않았다면 어떻게 되었을지도 갈아 끼워 볼 수 있다.
- 실행 순서
- —
- 총 소요
- —
- 결과
- —
$ dagster job execute -m atp --job atp_job atp_matches_dataset - STEP_START - Started execution of step "atp_matches_dataset". atp_players_dataset - STEP_START - Started execution of step "atp_players_dataset". atp_matches_dataset - STEP_SUCCESS - Finished in 413ms. atp_players_dataset - STEP_SUCCESS - Finished in 1.51s. atp_players_name_dataset - STEP_START - Started execution of step "atp_players_name_dataset". atp_players_name_dataset - STEP_SUCCESS - Finished in 49ms. $ duckdb atp.duckdb \ 'SELECT COLUMNS(col -> col LIKE "name_%") FROM players LIMIT 5'
| name_first | name_last | name_full |
|---|---|---|
| varchar | varchar | varchar |
| Gardnar | Mulloy | Gardnar Mulloy |
| Pancho | Segura | Pancho Segura |
| Frank | Sedgman | Frank Sedgman |
| Giuseppe | Merlo | Giuseppe Merlo |
| Richard | Gonzalez | Richard Gonzalez |
atp_matches_dataset과 atp_players_dataset은 둘 다 곧바로 시작하지만,
atp_players_name_dataset은 자기가 의존하는 atp_players_dataset이 끝난 뒤에야 시작한다.
선언 한 줄이 실행 그래프를 만든 것이다.
col LIKE "name_%"가 겹인용부호를 쓴다. 셸 명령 전체가 홑인용부호로 감싸여 있어
그 안에서 홑인용부호를 쓰기 어려웠던 것으로 보인다. 그러나 SQL에서 겹인용부호는 문자열이 아니라 식별자를 뜻한다.
제3장 3.5.1절은 같은 표현을 col LIKE 'valid%'로 홑인용부호로 썼다.
이식성을 생각하면 셸 인용을 달리 처리하는 편이 안전하다.
8.4.4
애셋 안에서 진전된 계산하기Advanced computation in assets
제6장에서 배운 대로 DuckDB는 pandas 데이터프레임을 질의할 수 있고, 그 능력을 Dagster 안에서도 쓸 수 있다. 다음 애셋은 테니스 대회 등급 메타데이터를 제공하는 데이터프레임을 수집한다.
저자들이 이 예제의 성격을 미리 밝혀둔다. 이것은 시연에 가깝지만, pandas가 제공하는 온갖 통계적·수치적 기능으로 파이썬에서 할 수 있는 실제 계산을 상상해보라는 것이다. DuckDB와 Dagster 애셋을 파이썬과 그 능력을 통해 통합할 수 있게 되면서, 복잡한 계산과 연산을 데이터 파이프라인의 일부로 수행하는 일까지 파이썬 개발자에게 가능하고 쉬운 일이 된다.
NOTE — 대회 등급
테니스 대회는 상금 규모와 걸린 랭킹 포인트로 범주가 나뉜다. 그랜드 슬램이 가장 값진 범주이고, 그다음이 투어 파이널스, 마스터스 1000, 그 밖의 투어 레벨, 챌린저, ITF 순이다.
import pandas as pd @asset def atp_levels_dataset(duckdb: DuckDBResource) -> None: levels_df = pd.DataFrame({ "short_name": [ "G", "M", "A", "C", "S", "F" ], "name": [ "Grand Slam", "Tour Finals", "Masters 1000s", "Other Tour Level", "Challengers", "ITFs" ], "rank": [ 5, 4, 3, 2, 1, 0 ] }) with duckdb.get_connection() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS levels AS SELECT * FROM levels_df """)
SQL 문자열 안의 levels_df가 파이썬 지역 변수 이름이다.
등록도 변환도 없이 데이터프레임을 테이블처럼 쓰는 제6장의 오리 시험이 여기서 다시 쓰인다.
"테이블처럼 보이면 질의할 수 있다"는 원칙이 파이프라인 코드 안에서 값을 한다.
다만 CREATE TABLE IF NOT EXISTS인 점을 눈여겨볼 만하다.
matches와 players는 CREATE OR REPLACE TABLE인데 이것만 다르다.
다시 실체화해도 이미 있으면 갱신되지 않는다. 정적 메타데이터이니 그것이 의도일 수 있으나,
등급 표를 고쳤을 때 반영되지 않는다는 뜻이기도 하다.
| short_name | name | rank |
|---|---|---|
| varchar | varchar | int64 |
| G | Grand Slam | 5 |
| M | Tour Finals | 4 |
| A | Masters 1000s | 3 |
| C | Other Tour Level | 2 |
| S | Challengers | 1 |
| F | ITFs | 0 |
duckdb atp.duckdb 'FROM levels'로 확인한 결과다. 데이터프레임 수집이 제대로 작동했다.
그런데 코드와 이름의 짝이 맞는가. 표시해둔 다섯 행이 문제다.
살펴본 김에 — 등급 코드와 이름의 짝이 어긋나 있다
NOTE는 등급을 값진 순서로 여섯 개 열거했다. 그랜드 슬램, 투어 파이널스, 마스터스 1000, 그 밖의 투어 레벨, 챌린저, ITF다. 그리고 목록 8.19는 코드를 G, M, A, C, S, F 순으로 놓고 그 이름들을 순서대로 그냥 짝지었다.
그런데 이 데이터셋—Jeff Sackmann의 tennis_atp—의 tourney_level 규약은 그 순서가 아니다. 저장소 문서가 밝히는 바로는 G는 그랜드 슬램, M은 마스터스 1000, A는 그 밖의 투어 레벨, C는 챌린저, S는 새틀라이트/ITF, F는 투어 파이널스다.
그러므로 G를 뺀 다섯 코드가 모두 엉뚱한 이름에 붙어 있다. 그리고 이 어긋남의 증거가 같은 책 안에 있다. 8.3.3절이 보여준 첫 경기의 tourney_level이 A이고 대회 이름이 United Cup인데, 이 표를 따르면 그것이 "Masters 1000s"가 된다. United Cup은 국가 대항 팀 대회이며 마스터스 1000이 아니다.
이 표는 뒤에서 경기 데이터와 조인해 등급 이름을 붙이는 데 쓰일 물건이니, 그대로 두면 보고서의 모든 등급 이름이 틀린다. 판본에 따라 규약이 달라질 수 있으니 데이터셋 저장소의 문서로 코드 대응을 확인한 뒤 쓰는 편이 안전하다.
8.4.5
MotherDuck으로 올리기Uploading to MotherDuck
Dagster 파이프라인이 필요한 데이터를 모두 성공적으로 수집했지만, 지금 DuckDB 데이터베이스는 내 기계에만 있다. 완전한 데이터 파이프라인이라면 애플리케이션이 클라우드에서 데이터를 쓸 수 있도록 데이터를 발행할 것이다. 제7장에서 배운 대로 로컬 데이터베이스를 MotherDuck에 발행하고 공유할 수도 있고, 데이터를 클라우드 인스턴스로 직접 실어 보낼 수도 있다. 여기서는 간결함을 위해 후자에 초점을 둔다.
# 목록 8.20 — 지금의 defs defs = Definitions( assets=all_assets, jobs=[atp_job], resources={"duckdb": DuckDBResource( database="atp.duckdb", )}, schedules=[atp_schedule], ) # 목록 8.21 — 저장 위치를 MotherDuck으로 바꾼다 import dotenv import os dotenv.load_dotenv() # CLI에서 돌릴 때만 필요하다. # Dagster UI는 .env에 정의된 속성을 알아서 집어 온다. mduck_token = os.getenv("motherduck_token") defs = Definitions( assets=all_assets, jobs=[atp_job], resources={"duckdb": DuckDBResource( database=f"md:md_atp_db?motherduck_token={mduck_token}", schema="main" # MotherDuck의 md_atp_db 데이터베이스에 접속한다 )}, schedules=[atp_schedule], )
제7장 7.3.7절이 한 문장으로 정리한 그대로다. 접속 문자열에 md: 접두사를 끼워 넣고
?motherduck_token=<token> 매개변수를 덧붙이는 것뿐이다.
애셋 코드는 한 줄도 고치지 않았다. 리소스 정의만 바뀌었고, 애셋들은 자기가 어디에 쓰는지 모른다.
그것이 리소스를 주입하는 설계의 값어치다.
# 토큰을 셸에 내보낸 다음 작업을 돌린다. $ dagster job execute -m atp --job atp_job
〈도판 8.8〉 · 화면 갈무리
작업이 끝나면 MotherDuck UI에서 md_atp_db 데이터베이스가 만들어졌음을 확인할 수 있다. 원서의 이 도판은 그 화면의 갈무리다. 재현하지 않고 서술로 대신한다.
저자들의 맺음말이 이 장을 다음 장으로 잇는다.
Dagster가 데이터 적재와 변환을 조율했고, 추가 정보의 계산까지 했다.
마지막 걸음으로 결과를 클라우드 데이터 저장소에 발행해, 그 위에 제품을 지을 준비를 마쳤다.
그 제품을 짓는 것이 다음 장의 일이다.
| 도구 | 맡는 자리 | 쓰는 언어 | DuckDB의 역할 | 이 장에서 만든 것 |
|---|---|---|---|---|
| dlt | 수집 ingest | Python | 목적지 — 저장소로 쓰인다 | chess.com API → chess_pipeline.duckdb |
| dbt | 변환 transform | YAML + SQL | 저장소 아님 — 변환 수단으로만 쓰인다 | CSV 55개 → matches.parquet |
| Dagster | 조율 orchestrate | Python | 목적지 — 로컬 또는 MotherDuck | CSV 56개 + 데이터프레임 → atp.duckdb → md_atp_db |
표를 옆으로 읽으면 이 장의 구조가 보인다. 같은 데이터베이스가 세 도구에서 세 가지 자리에 앉는다. dlt에서는 데이터가 도착하는 목적지이고, dbt에서는 들어오지도 나가지도 않고 가운데에서 계산만 하는 엔진이며, Dagster에서는 다시 목적지이면서 그 목적지의 위치를 설정 한 줄로 클라우드로 옮길 수 있는 대상이다. 8.1절이 "대개 변환 부분에 앉는다"고 한 말은 규칙이 아니라 경향의 서술이었던 것이다.
SUMMARY
제8장이 남긴 여덟 문장Summary
- 데이터 파이프라인은 자동화되고 일관된 방식으로 데이터를 적재하고 변환하게 해준다.
- DuckDB는 dlt, dbt, Dagster 같은 여러 데이터 파이프라인 도구에 손쉽게 통합될 수 있다.
- DuckDB는 파이프라인에서 데이터 적재, 변환, 저장이라는 여러 역할을 할 수 있다.
- 데이터 변환과 필터링은 SQL과 파이썬 API 양쪽으로 할 수 있다.
- 데이터는 CSV 파일 한 벌(그 밖의 원천과 형식도)과 pandas 데이터프레임에서 원격으로 적재할 수 있다.
- 파이프라인에 파이썬을 쓰면 강력한 변환과 계산이 가능해진다.
- 의존을 선언하는 것이 파이프라인 안 연산의 순서를 조율하는 데 도움이 된다.
- MotherDuck은 클라우드에 둘 DuckDB 데이터의 알맞은 목적지다.