Pandas로 운석 데이터 처리하기

Pandas 구현은 단일 노드의 메모리에 데이터를 읽습니다. 운석 입력을 준비하고 Pandas를 환경에 추가한 뒤 ROOT를 실제 프로젝트 폴더로 바꿉니다.

필터, 평균과 비교

src/myproject/datasets/meteor_analysis.py:

from transforms.api import Input, Output, transform

ROOT = "/Example Project/Meteorites"


@transform.using(
    output=Output(f"{ROOT}/meteorite_landings_cleaned"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
)
def clean(output, meteorite_landings):
    rows = meteorite_landings.pandas()
    selected = rows.loc[
        (rows["nametype"] == "Valid") & (rows["year"] >= 1950)
    ]
    output.write_table(selected)


@transform.using(
    output=Output(f"{ROOT}/meteorite_stats"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
)
def stats(output, meteorite_landings):
    averages = (
        meteorite_landings.pandas()
        .groupby("class", as_index=False)["mass"]
        .mean()
        .rename(columns={"mass": "avg_mass_per_class"})
    )
    output.write_table(averages)


@transform.using(
    output=Output(f"{ROOT}/meteorite_enriched"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
    meteorite_stats=Input(f"{ROOT}/meteorite_stats"),
)
def enriched(output, meteorite_landings, meteorite_stats):
    landings = meteorite_landings.pandas()
    averages = meteorite_stats.pandas()
    joined = landings.merge(averages, on="class", how="inner")
    joined["greater_mass"] = joined["mass"] > joined["avg_mass_per_class"]
    output.write_table(joined.loc[joined["greater_mass"]])

모든 함수는 Pandas 핸들과 write_table()을 사용하므로 단일 노드 실행으로 구성합니다.

Clean의 결과와 별도로 Stats 및 Enriched는 원본을 기준으로 계산합니다. Pandas의 groupby는 기본적으로 null 그룹 키를 제외하므로 다른 데이터나 엔진으로 일반화할 때 이 정책을 확인합니다.

Class별 출력 생성

src/myproject/datasets/meteor_class.py:

from transforms.api import Input, Output, transform

ROOT = "/Example Project/Meteorites"


def make_class_transform(class_name):
    @transform.using(
        output=Output(f"{ROOT}/meteorite_{class_name}"),
        my_input=Input(f"{ROOT}/meteorite_enriched"),
    )
    def filter_by_class(output, my_input):
        rows = my_input.pandas()
        output.write_table(rows.loc[rows["class"] == class_name])

    filter_by_class.__name__ = f"filter_{class_name}"
    return filter_by_class


TRANSFORMS = [make_class_transform(name) for name in ["L6", "H5", "H4"]]

각 변환은 같은 Enriched 입력에서 자신의 class만 선택합니다. 세 출력은 별도 변환이므로 빌드마다 원본 입력을 다시 읽을 수 있습니다. 대용량이거나 공통 계산이 비싸면 엔진 및 다중 출력의 실행 비용을 함께 검토합니다.