스트리밍 (Streaming)

스트리밍 (Streaming)

레이지(lazy) API가 주는 이점 중 하나가 또 있는데요, 바로 쿼리를 스트리밍 방식으로 실행할 수 있다는 점이에요. 한 번에 모든 데이터를 처리하는 대신, Polars가 쿼리를 배치(batch) 단위로 나눠서 실행하기 때문에 메모리에 다 들어가지 않는 데이터셋도 처리할 수 있답니다. 메모리 부담을 줄여주는 것뿐 아니라, 스트리밍 엔진은 Polars의 인메모리 엔진보다 성능도 더 좋아요.

쿼리를 스트리밍 모드로 실행하고 싶다면, collect에 engine="streaming" 인자를 넘겨주면 돼요.

q1 = (
    pl.scan_csv("docs/assets/data/iris.csv")
    .filter(pl.col("sepal_length") > 5)
    .group_by("species")
    .agg(pl.col("sepal_width").mean())
)
df = q1.collect(engine="streaming")

let q1 = LazyCsvReader::new(PlRefPath::new("docs/assets/data/iris.csv"))
.with_has_header(true)
.finish()?
.filter(col("sepal_length").gt(lit(5)))
.group_by(vec![col("species")])
.agg([col("sepal_width").mean()]);
let df = q1.clone().with_streaming(true).collect()?;
println!("{df}");

스트리밍 쿼리 확인하기

Polars는 많은 연산을 스트리밍 방식으로 실행할 수 있어요. 다만 어떤 연산은 원래 스트리밍이 불가능하거나 아직 스트리밍으로 구현되어 있지 않을 수도 있고요. 그런 경우에는 Polars가 그 연산에 대해서만 인메모리 엔진으로 대체해서 실행해요.

q1 = (
    pl.scan_csv("docs/assets/data/iris.csv")
    .filter(pl.col("sepal_length") > 5)
    .group_by("species")
    .agg(
        mean_width=pl.col("sepal_width").mean(),
        mean_width2=pl.col("sepal_width").sum() / pl.col("sepal_length").count(),
    )
    .show_graph(plan_stage="physical", engine="streaming")
)