Polars로 운석 데이터 처리하기

운석 입력을 준비하고 환경에 Polars를 추가합니다. 다음의 ROOT를 실제 프로젝트 폴더로 바꿉니다. 이 구현은 단일 노드에서 실행하며 수치형 year와 mass, 문자열 nametype과 class를 사용합니다.

필터, 평균과 비교

src/myproject/datasets/meteor_analysis.py:

import polars as pl
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.polars().filter(
        (pl.col("nametype") == "Valid") & (pl.col("year") >= 1950)
    )
    output.write_table(rows)


@transform.using(
    output=Output(f"{ROOT}/meteorite_stats"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
)
def stats(output, meteorite_landings):
    averages = (
        meteorite_landings.polars()
        .group_by("class")
        .agg(pl.col("mass").mean().alias("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.polars()
    averages = meteorite_stats.polars()
    joined = landings.join(averages, on="class", how="inner")
    compared = joined.with_columns(
        (pl.col("mass") > pl.col("avg_mass_per_class")).alias("greater_mass")
    )
    output.write_table(compared.filter(pl.col("greater_mass")))

Clean은 Valid 및 1950 이상 조건을 적용합니다. Stats와 Enriched는 원본 전체를 사용하여 class별 평균보다 큰 질량을 찾습니다. 평균과 같은 값은 엄격한 > 조건을 통과하지 않습니다.

Class별 출력 생성

src/myproject/datasets/meteor_class.py:

import polars as pl
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):
        selected = my_input.polars().filter(pl.col("class") == class_name)
        output.write_table(selected)

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


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

Factory 호출이 각 class 값을 고정하고 리스트에 세 변환을 보관합니다. 기본 datasets 패키지의 자동 등록으로 발견되며 출력은 L6, H5, H4별로 분리됩니다.

다른 데이터로 확장할 때는 class가 null인 행의 집계 및 조인 정책도 명시해야 합니다. 같은 키 이름을 사용하더라도 엔진별 null 처리까지 자동으로 같아지는 것은 아닙니다.