Spark로 운석 데이터 처리하기

분산 실행을 사용하려면 @transform.spark.using을 선택합니다. 운석 입력의 스키마를 준비하고 ROOT를 실제 프로젝트 폴더로 바꿉니다. 이 구문에는 transforms 3.95.0 및 해당 템플릿의 지원 버전이 필요합니다.

필터, 평균과 비교

src/myproject/datasets/meteor_analysis.py:

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform

ROOT = "/Example Project/Meteorites"


@transform.spark.using(
    output=Output(f"{ROOT}/meteorite_landings_cleaned"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
)
def clean(output, meteorite_landings):
    rows = meteorite_landings.dataframe()
    selected = rows.filter(
        (F.col("nametype") == "Valid") & (F.col("year") >= 1950)
    )
    output.write_dataframe(selected)


@transform.spark.using(
    output=Output(f"{ROOT}/meteorite_stats"),
    meteorite_landings=Input(f"{ROOT}/meteorite_landings"),
)
def stats(output, meteorite_landings):
    averages = meteorite_landings.dataframe().groupBy("class").agg(
        F.mean("mass").alias("avg_mass_per_class")
    )
    output.write_dataframe(averages)


@transform.spark.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.dataframe()
    averages = meteorite_stats.dataframe()
    joined = landings.join(averages, on="class", how="inner")
    compared = joined.withColumn(
        "greater_mass", F.col("mass") > F.col("avg_mass_per_class")
    )
    output.write_dataframe(compared.filter(F.col("greater_mass")))

dataframe()으로 받은 표에 열 표현식을 적용합니다. Stats와 Enriched의 모집단은 원본 전체입니다.

Class별 출력 생성

src/myproject/datasets/meteor_class.py:

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform

ROOT = "/Example Project/Meteorites"


def make_class_transform(class_name):
    @transform.spark.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.dataframe().filter(F.col("class") == class_name)
        output.write_dataframe(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 값을 보관하고 세 변환을 모듈 리스트에 등록합니다. 이 경로는 Spark 출력이므로 write_dataframe()을 사용합니다. 완료한 Enriched 데이터셋을 Contour 같은 분석 도구에서 살펴볼 수 있습니다.