코드 노드로 API 응답을 데이터셋에 적재하기

파이프라인은 컬렉션의 데이터셋, 코드, 변환 노드를 캔버스에서 연결해 실행하는 작업 흐름입니다.

이번 레슨에서는 첫 파이프라인을 만듭니다. 입력 데이터셋은 없습니다. 데이터를 만들어 내는 쪽이 외부 API이므로, 커넥터를 바인딩한 코드 노드가 파이프라인의 출발점이 됩니다.

코드가 필요한 이유

레슨 02에서 확인한 응답은 컬럼별 배열을 담은 중첩 객체였습니다.

hourly.time[]                  ["2026-08-03T00:00", "2026-08-03T01:00", ...]
hourly.temperature_2m[]        [27.6, 27.1, ...]
hourly.relative_humidity_2m[]  [87, 89, ...]

데이터셋은 행의 집합이므로 이 세 배열을 같은 인덱스끼리 묶어 한 행으로 만들어야 합니다. 이 처리는 이름변경이나 집계 같은 표준 변환으로 표현할 수 없습니다. 코드 노드를 사용하는 이유입니다.

결과 데이터셋 먼저 만들기

파이프라인 편집기를 열기 전에 컬렉션에 src_weather_hourly 테이블을 만듭니다. 빠른 추가로 만든 출력 테이블을 연결하면 저장 과정에서 데이터셋 참조 오류가 발생할 수 있기 때문입니다.

1. 왼쪽 사이드바에서 컬렉션을 선택하고 실습 컬렉션을 엽니다.
2. 항목 추가를 선택하고 데이터셋에 마우스를 올린 뒤 테이블을 선택합니다.
3. 기본 정보에서 이름을 src_weather_hourly로 입력합니다. 목록에서도 같은 이름이 보이도록 별칭은 비워 두거나 src_weather_hourly로 입력한 뒤 다음을 선택합니다.

4. 스키마에서 다음 네 컬럼을 구성합니다.

5. 기본 field1을 실제 컬럼 이름으로 바꾸고 나머지 세 컬럼을 추가하면 하단 버튼이 생성으로 바뀝니다. 생성을 선택합니다.

컬럼 이름 두 개를 API가 준 그대로 둔 것은 의도적입니다. 원본 계층은 API의 표현을 보존하고, 읽기 좋은 이름은 다음 레슨의 정리 코드가 만듭니다.

Python 코드 자산 만들기

파이프라인을 열기 전에 코드도 컬렉션 자산으로 먼저 만듭니다. 빠른 추가로 만든 임시 코드 노드는 저장할 때 유효한 코드 ID로 해석되지 않을 수 있으므로, 이 과정에서는 사용하지 않습니다.

1. 실습 컬렉션에서 항목 추가 → 코드 → Python을 선택합니다.
2. 이름과 별칭을 collect_weather_hourly로 입력합니다.
3. 설명은 Open-Meteo에서 시간별 날씨 수집으로 입력합니다.
4. 다음을 선택합니다. 아래 코드는 기본 런타임의 polars를 사용하므로 추가 패키지를 선언하지 않습니다.

커넥터 바인딩하기

코드에서 커넥터를 쓰려면 @use_connector 데코레이터로 선언해야 합니다. 직접 타이핑하지 않고 편집기가 넣어 주도록 합니다.

1. 코드 편집기의 첫 줄에서 @를 입력합니다.
2. 나타나는 목록에서 connector를 선택합니다.
3. 커넥터 목록에서 open_meteo를 선택합니다.

다음 한 줄이 삽입됩니다.

@use_connector("open_meteo", ref="open_meteo")

첫 번째 인자는 코드 안에서 사용할 변수 이름이고, ref는 참조할 커넥터 이름입니다. 실행 시점에 런타임이 이 이름으로 커넥터를 찾아 연결 객체를 만들어 변수에 넣어 줍니다. 커넥터에 저장된 주소와 자격 증명은 코드에 노출되지 않습니다.

코드 작성하기

데코레이터 아래에 다음 코드를 작성하고 생성을 선택합니다.

@use_connector("open_meteo", ref="open_meteo")
def run(options=None, contexts=None):
    import polars as pl

    body = open_meteo.get(  # noqa: F821 — @use_connector가 주입합니다
        "/v1/forecast",
        {
            "latitude": 37.5665,
            "longitude": 126.9780,
            "hourly": "temperature_2m,relative_humidity_2m",
            "past_days": 3,
            "forecast_days": 0,
            "timezone": "Asia/Seoul",
        },
    )

    hourly = body["hourly"]
    frame = pl.DataFrame(
        {
            "observed_at": hourly["time"],
            "observed_date": [value[:10] for value in hourly["time"]],
            "temperature_2m": hourly["temperature_2m"],
            "relative_humidity_2m": hourly["relative_humidity_2m"],
        },
        schema_overrides={
            "observed_at": pl.String,
            "observed_date": pl.String,
            "temperature_2m": pl.Float64,
            "relative_humidity_2m": pl.Int64,
        },
    )
    return {"src_weather_hourly": frame}

생성이 끝나면 컬렉션 목록에서 collect_weather_hourly를 열고 코드 탭에 방금 작성한 소스가 보이는지 확인합니다. 코드 이름만 생기고 소스가 비어 있으면 편집 → 코드에서 다시 붙여넣고 변경사항 저장을 선택합니다. 소스 파일이 없는 코드는 파이프라인 저장 또는 실행 단계에서 실패합니다.

코드에서 확인할 점이 세 가지 있습니다.

• run에 입력 인자가 없습니다. 입력 데이터셋이 연결되지 않은 코드 노드는 인자 없이 호출됩니다. 데이터의 출처는 커넥터뿐입니다.
• get은 파싱된 JSON을 돌려줍니다. 응답 객체가 아니라 딕셔너리이므로 body["hourly"]로 바로 접근합니다. 경로는 커넥터의 Base URL 뒤에 붙습니다.
• 반환 딕셔너리의 키가 출력 포트 이름입니다. src_weather_hourly는 잠시 뒤 연결할 출력 데이터셋 이름과 정확히 같아야 합니다.

observed_date는 문자열 앞 10글자를 잘라 만듭니다. 레슨 04의 집계가 이 컬럼을 기준으로 하루씩 묶습니다.

반환하는 frame은 Polars DataFrame입니다. schema_overrides로 데이터셋의 네 컬럼 타입을 지정하므로 응답이 비었거나 관측값에 null이 있어도 타입이 유지됩니다. 다음 레슨의 코드 노드도 테이블 입력을 Polars로 받습니다.

편집기를 열고 컬렉션 선택하기

1. 왼쪽 사이드바에서 데이터 → 파이프라인을 선택합니다.
2. 오른쪽 위의 생성을 선택합니다.
3. 파이프라인 컬렉션 선택 화면에서 실습 컬렉션을 선택합니다. 선택하면 별도의 계속 클릭 없이 편집기가 열립니다.

컬렉션을 고르기 전에는 항목을 추가하거나 실행할 수 없습니다. 하나의 파이프라인에는 선택한 컬렉션 안의 자산만 사용할 수 있습니다.

출력 데이터셋 연결하기

1. 컴포넌트 검색란에서 collect_weather_hourly를 찾습니다. 항목 오른쪽의 세로 점 모양 드래그 핸들을 잡아 빈 캔버스로 끌어다 놓습니다.
2. 같은 방법으로 src_weather_hourly를 검색해 캔버스로 끌어다 놓습니다. 이름을 잡으면 선택만 될 수 있으므로 오른쪽 드래그 핸들을 사용합니다.
3. 코드 노드의 오른쪽 연결점을 src_weather_hourly의 왼쪽 연결점에 연결합니다.

4. 코드 노드를 선택하고 검사기의 옵션 탭을 엽니다.
5. 출력 (1) 아래의 src_weather_hourly 행을 펼칩니다.
6. 반환 키와 데이터셋을 확인하고 쓰기 모드를 전체 교체로 바꾼 뒤 검사기 아래의 저장을 선택합니다.

collect_weather_hourly → src_weather_hourly

출력 포트 이름이 반환 딕셔너리의 키와 다르면 실행이 실패합니다. 검사기의 출력에 표시된 이름과 코드의 return {"...": frame} 키를 나란히 비교하세요.

저장하고 한 번 실행하기

1. 오른쪽 위의 저장을 선택합니다.
2. 이름을 weather_daily_pipeline으로 입력하고 파이프라인 유형은 배치로 둡니다.
3. 저장이 끝나면 지금 실행을 선택합니다.

실행이 끝나면 컬렉션에서 src_weather_hourly를 열고 데이터 탭에서 다음을 확인합니다.

• 전체 행이 72개인지. 24시간 × 3일입니다.
• observed_date에 서로 다른 날짜가 3개 있는지.
• temperature_2m에 소수점이 있는 값이, relative_humidity_2m에 정수가 들어 있는지.

자가 점검

• 컬렉션에 네 컬럼 스키마의 src_weather_hourly를 먼저 만들었는지 확인합니다.
• 코드 노드에 @use_connector("open_meteo", ref="open_meteo")가 선언됐는지 확인합니다.
• 반환 딕셔너리의 키와 출력 포트 이름이 모두 src_weather_hourly인지 확인합니다.
• 쓰기 방식이 전체 교체인지 확인합니다.
• 실행 후 데이터셋에 72행이 적재됐는지 확인합니다.

다음 레슨

같은 파이프라인에 정리·집계 코드 노드를 이어 붙입니다. API가 준 컬럼 이름을 읽기 좋게 바꾸고, 시간별 72행을 일별 3행으로 집계해 분석용 마트를 만듭니다.