컬렉션 코드로 Python과 SQL 변환 작성하기

기본 변환으로 표현하기 어려운 로직은 Python 또는 SQL 코드 자산으로 작성합니다. 현재 편집기의 코드 노드는 파이프라인 안에만 저장되는 인라인 스크립트가 아닙니다. 선택한 컬렉션에 저장된 Code 자산을 참조합니다.

이번 레슨에서는 Python 코드 자산을 만들고 이전 레슨의 입력 데이터셋을 처리합니다. 레슨 3의 결과는 그대로 두고, 코드 실행 결과를 저장할 새 데이터셋과 파이프라인을 만듭니다.

코드는 언제 사용할까요?

컬럼 선택, 필터, 이름 변경, 형변환, 조인, 유니온, 집계는 먼저 변환 그룹의 기본 노드를 검토하세요. 다음과 같은 로직은 코드가 더 명확할 수 있습니다.

• 정규식으로 "4,660₩"을 숫자와 통화 코드로 나누기
• "Seoul", "서울", "SEL"처럼 서로 다른 값을 하나로 표준화하기
• 외부 코드표를 조회해 새 컬럼을 추가하기
• 한 행을 여러 행으로 펼치거나 복잡한 품질 규칙을 검사하기

코드가 하는 일

이번 실습에서는 주문 금액이 10,000 이상인 행만 남기는 조건을 Python으로 작성합니다. 실행 전후의 행 수와 금액을 비교하면 코드가 적용됐는지 바로 확인할 수 있습니다.

이 조건 하나만 적용한다면 기본 선택/필터 변환이 더 간단합니다. 여기서는 같은 필터를 Python으로 작성하면서 코드 노드의 입력, 반환 형식, 연결 방법을 익히는 것이 목적입니다. 실제 업무에서는 여러 조건이나 사용자 정의 계산처럼 기본 변환만으로 표현하기 어려운 규칙을 코드로 작성하고 재사용합니다.

코드 결과를 저장할 데이터셋 만들기

이미 데이터가 적재된 레슨 3의 출력 데이터셋은 변경하지 않습니다. 다음 순서로 필터링 결과 전용 데이터셋을 새로 만듭니다.

1. 왼쪽 사이드바에서 컬렉션을 선택하고 src_postgres_orders가 있는 컬렉션을 엽니다.
2. 항목 추가를 선택하고 데이터셋에 마우스를 올린 뒤 테이블을 선택합니다.
3. 기본 정보에서 이름을 mart_orders_filtered로 입력하고 다음을 선택합니다.
4. 스키마에서 임시 컬럼을 수정하고 속성 추가로 다음 세 컬럼을 구성합니다.

5. 생성을 선택하고 컬렉션 항목 목록에 mart_orders_filtered가 표시되는지 확인합니다. 데이터셋을 연 직후에는 스키마만 있고 데이터는 0행이어야 합니다.

이미 같은 이름의 데이터셋이 있다면 스키마가 위 구성과 같은지 확인하고 기존 데이터셋을 사용해도 됩니다. 실행 전부터 결과가 들어 있다면 실행 이력도 함께 확인해야 이번 실행의 성공 여부를 정확히 판단할 수 있습니다.

새 파이프라인 열기

1. 왼쪽 사이드바에서 데이터 → 파이프라인을 선택합니다.
2. 오른쪽 위의 생성을 선택합니다.
3. 파이프라인 컬렉션 선택 창에서 src_postgres_orders와 mart_orders_filtered가 있는 컬렉션을 선택합니다.

레슨 3의 파이프라인을 수정하지 않고 새 파이프라인을 만드는 이유는 두 실습 결과를 각각 보존하고 실행 결과를 혼동하지 않기 위해서입니다.

Python 코드 자산 만들기

컬렉션이 선택된 상태에서 다음 순서로 진행합니다.

1. 컴포넌트 라이브러리에서 빠른 추가 → 코드 → Python을 캔버스로 끌어다 놓습니다.
2. 코드 생성 창에서 이름을 filter_high_value_orders로 입력합니다.
3. 아래 코드를 작성합니다.

import polars as pl

def run(src_postgres_orders, *, options=None, contexts=None):
    output = (
        src_postgres_orders
        .with_columns(
            pl.col("amount").cast(pl.Int64),
            pl.col("ord_dt_str").alias("order_dt_str"),
        )
        .filter(pl.col("amount") >= 10000)
        .select(["order_id", "order_dt_str", "amount"])
    )
    return {"mart_orders_filtered": output}

파이프라인 런타임은 연결 이름을 키워드 인자로 사용해 코드를 호출합니다. 따라서 입력 포트가 src_postgres_orders라면 함수 인자도 src_postgres_orders여야 합니다. 반환 딕셔너리의 mart_orders_filtered도 출력 포트 이름과 정확히 같아야 합니다. 둘 중 하나가 다르면 unexpected keyword argument 또는 출력 키 관련 오류가 발생합니다.

새 Python 항목은 파이프라인을 저장할 때 선택한 컬렉션의 Code 자산으로 함께 저장됩니다. 다른 파이프라인에서도 재사용할 수 있으므로 파이프라인 전용 임시 코드라고 생각하면 안 됩니다.

데이터셋과 코드 연결하기

1. 컴포넌트 라이브러리의 컬렉션 영역에서 src_postgres_orders를 캔버스로 끌어다 놓습니다.
2. src_postgres_orders의 오른쪽 연결점을 filter_high_value_orders 코드 노드의 왼쪽 연결점에 연결합니다.
3. 컬렉션 영역에서 앞에서 만든 mart_orders_filtered를 캔버스로 끌어다 놓습니다.
4. 코드 노드의 오른쪽 연결점을 mart_orders_filtered에 연결합니다.
5. 코드 노드를 선택하고 검사기의 옵션 탭을 엽니다.
6. 입력에 src_postgres_orders가 표시되는지 확인하고, 이 이름이 코드의 run 함수 인자와 정확히 같은지 비교합니다. 읽기 방식은 증분으로 둡니다.
7. 출력에 mart_orders_filtered가 표시되고 쓰기 방식이 추가인지 확인합니다.

src_postgres_orders → filter_high_value_orders → mart_orders_filtered

캔버스에는 src_postgres_orders, filter_high_value_orders, mart_orders_filtered 세 노드가 있어야 합니다. 레슨 3의 mart_orders_clean은 그대로 보존되며, 이번 코드 실행 결과는 새 데이터셋에 따로 저장됩니다.

requires exactly one output (got 0) 오류가 표시되면 코드 또는 변환 노드의 출력이 연결되지 않았다는 뜻입니다. 해당 노드의 오른쪽 연결점에서 mart_orders_filtered까지 연결한 뒤 다시 저장하세요.

기존 코드 자산 재사용하기

이미 컬렉션에 코드가 있다면 새로 만들 필요가 없습니다.

• 빠른 추가 → 코드 → 기존 코드를 캔버스로 끌어다 놓고 목록에서 선택합니다.
• 또는 컴포넌트 라이브러리의 컬렉션 영역을 펼쳐 해당 코드 자산을 캔버스로 끌어다 놓습니다.

재사용 중인 코드 자산을 수정하면 그 코드를 참조하는 다른 파이프라인에도 다음 실행부터 변경 내용이 적용될 수 있습니다. 검사기에서 저장할 때 원본 덮어쓰기와 새 코드로 저장 중 목적에 맞는 방법을 선택하세요.

저장하고 실행하기

1. 저장을 선택하고 파이프라인 이름을 mart_orders_filtered_pipeline으로 입력합니다.
2. 지금 실행을 선택합니다.
3. 실행 이력에서 실행이 완료됐는지 확인합니다.
4. 컬렉션에서 mart_orders_filtered를 열고 데이터 탭을 확인합니다.

정상적으로 실행되면 금액이 10,000 이상인 주문 두 행만 확인할 수 있습니다. 금액이 8,900인 order_002는 결과에서 제외됩니다.

자가 점검

• 새 Python 또는 SQL 항목이 선택한 컬렉션의 Code 자산으로 저장됐는지 확인합니다.
• 데이터셋 → 코드 → 데이터셋 순서로 연결됐는지 확인합니다.
• 검사기의 옵션에서 입력과 출력 데이터셋이 올바르게 연결됐는지 확인합니다.
• 실행 이력에서 이번 실행이 완료됐는지 확인합니다.
• 실행 후 mart_orders_filtered에서 금액이 10,000 이상인 두 행만 확인합니다.

다음 레슨

다음 레슨에서는 배치 파이프라인을 정해진 시간에 실행하도록 일정을 설정합니다.