파이프라인으로 엔티티 데이터 적재하기

이전 레슨에서는 iot_machine의 구조만 정의했습니다. 이번에는 원본 데이터셋을 만들고, 배치 파이프라인의 Entity 출력으로 두 행을 적재합니다. 엔티티와 데이터셋을 같은 컬렉션에 두면 컴포넌트 라이브러리에서 함께 선택할 수 있습니다.

사전 준비

• 이전 레슨에서 만든 iot_machine 엔티티
• iot_machine의 Identity Keys로 지정한 id
• 엔티티와 파이프라인을 만들 수 있는 컬렉션 편집 권한

machines 원본 데이터셋 만들기

1. 왼쪽 사이드바에서 컬렉션을 선택하고 실습 컬렉션을 엽니다.
2. 항목 추가 → 데이터셋 → 테이블을 선택합니다.
3. 기본 정보에서 이름을 machines로 입력하고 다음을 선택합니다.
4. 스키마에서 다음 컬럼을 만듭니다.

5. 생성을 선택합니다.

6. machines.csv 내려받기를 클릭해 준비된 파일을 저장합니다.

7. machines를 열고 데이터 → 업로드에서 CSV를 추가합니다.

8. 업로드가 끝나면 데이터 탭에 2행이 보이는지 확인합니다.

원본의 latest_recorded_at은 Text지만 엔티티 속성은 Timestamp입니다. 파이프라인 Code에서 이 타입 차이를 변환합니다.

데이터셋과 엔티티의 NULL 허용 설정이 반드시 같아야 하는 것은 아닙니다. 다만 원본에 NULL이 있을 수 있는데 대상 속성이 NOT NULL이면 적재가 실패할 수 있습니다. 이번 실습에서는 앞 레슨의 엔티티 계약과 맞춰 id만 NOT NULL로 두고 나머지 컬럼은 NULL을 허용합니다.

배치 파이프라인 만들기

1. 왼쪽 사이드바에서 데이터 → 파이프라인을 엽니다.
2. 오른쪽 위 생성을 선택합니다.
3. 파이프라인 컬렉션 선택에서 machines와 iot_machine이 있는 컬렉션을 선택합니다.
4. 파이프라인 유형은 배치로 유지합니다.

Python Code 만들기

1. 컴포넌트 라이브러리에서 빠른 추가 → 코드 → Python을 캔버스로 끌어다 놓습니다.
2. 이름을 map_machines_to_iot_machine으로 입력합니다.
3. 다음 코드를 입력합니다.

import polars as pl

def run(machines, *, options=None, contexts=None):
    output = machines.with_columns(
        pl.col("id").cast(pl.Utf8),
        pl.col("sensor_count").cast(pl.Int32),
        pl.col("reading_count").cast(pl.Int32),
        pl.col("latest_recorded_at").str.to_datetime(strict=False),
        pl.col("anomaly_count").cast(pl.Int32),
        pl.col("latest_health_score").cast(pl.Float64),
    )
    return {"iot_machine": output}

런타임은 파이프라인 입력 키를 run 함수의 키워드 인자로 전달합니다. 따라서 입력 키 machines와 함수 인자 machines가 같아야 합니다. 반환 딕셔너리의 키 iot_machine도 출력 키와 같아야 합니다.

입력·Code·Entity 연결하기

1. 컴포넌트 라이브러리의 컬렉션 영역에서 machines를 캔버스에 추가합니다.
2. machines의 오른쪽 연결점을 Code 노드의 왼쪽 연결점에 연결합니다.
3. 빠른 추가 → 온톨로지 → 엔티티를 캔버스로 끌어다 놓습니다.
4. 엔티티 소스 추가 창에서 별칭이 IoT 머신인 iot_machine을 선택하고 추가를 선택합니다.
5. Code 노드의 오른쪽 연결점을 iot_machine에 연결합니다.
6. Code 노드를 선택하고 옵션에서 입력 키가 machines, 출력 키가 iot_machine인지 확인합니다.
7. 엔티티 출력의 쓰기 방식이 병합인지 확인합니다.

컬렉션 영역에는 기존 데이터셋과 코드만 표시되므로 엔티티는 나타나지 않습니다. 기존 엔티티와 관계는 빠른 추가의 온톨로지 항목을 통해 선택합니다.

flowchart LR
    accTitle: 데이터셋에서 엔티티로 적재하는 파이프라인
    accDescr: machines 원본 데이터셋을 Python Code가 엔티티 스키마로 변환하고 iot_machine 엔티티에 병합하는 흐름입니다.
    dataset[(machines<br/>원본 데이터셋)] -->|입력 키 machines| code[map_machines_to_iot_machine<br/>Python Code]
    code -->|출력 키 iot_machine| entity[(iot_machine<br/>Entity 병합)]

저장하고 실행하기

1. 저장을 선택하고 파이프라인 이름을 load_iot_machines로 입력합니다.
2. 지금 실행을 선택합니다.
3. 상단 상태가 준비됨, 최근 실행이 성공으로 표시될 때까지 기다립니다.
4. 실행 상세에서 map_machines_to_iot_machine 스텝도 성공했는지 확인합니다.

병합은 같은 id가 다시 들어오면 기존 행을 갱신합니다. 이 파이프라인을 다시 실행해도 machine_001, machine_002가 중복 행으로 늘어나지 않아야 합니다.

데이터 탭에서 결과 확인하기

1. 온톨로지 → 모델링으로 돌아갑니다.
2. 실습 컬렉션에서 IoT 머신 노드를 선택합니다.
3. 데이터 탭을 열고 필요하면 새로 고칩니다.
4. 다음 두 행의 id와 변환된 속성을 확인합니다.

그래프 반영은 백그라운드에서 처리되므로 그래프 탐색기에는 잠시 뒤 나타날 수 있습니다.

자가 점검

• machines 데이터셋에 2행을 업로드했습니다.
• 입력 키와 함수 인자가 모두 machines입니다.
• 반환 키와 Entity 출력 키가 모두 iot_machine입니다.
• 쓰기 방식이 병합입니다.
• 실행 상태 준비됨과 최근 실행 성공을 확인했습니다.
• iot_machine의 데이터 탭에 두 행이 표시됩니다.

다음 레슨

다음 레슨에서는 iot_sensor와 iot_machine 사이의 관계를 정의하고, 관계 구조 컬럼을 파이프라인의 Relation 출력으로 적재합니다.