Skip to content

Migrate from Spark

A PySpark batch job that reads files from object storage, transforms them with the DataFrame API and writes files back runs on Polars Nanook after rewriting the transformation against the Polars LazyFrame API and submitting it with .remote(). The same job in both APIs:

# PySpark
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.getOrCreate()

customer = spark.read.parquet("s3://<BUCKET>/customer/")
orders = spark.read.parquet("s3://<BUCKET>/orders/")

result = (
    customer.filter(F.col("c_mktsegment") == "BUILDING")
    .join(orders, customer.c_custkey == orders.o_custkey)
    .groupBy("o_orderkey")
    .agg(F.sum("o_totalprice").alias("revenue"))
)
result.write.mode("overwrite").parquet("s3://<BUCKET>/revenue/")

import polars as pl
import polars_nanook as pn

ctx: pn.ComputeContext | pn.ClusterContext

customer = pl.scan_parquet("s3://<BUCKET>/customer/*.parquet")
orders = pl.scan_parquet("s3://<BUCKET>/orders/*.parquet")

result = (
    customer.filter(pl.col("c_mktsegment") == "BUILDING")
    .join(orders, left_on="c_custkey", right_on="o_custkey")
    .group_by("o_orderkey")
    .agg(pl.sum("o_totalprice").alias("revenue"))
)
result.remote(ctx).sink_parquet("s3://<BUCKET>/revenue/").await_result()

What changed:

  • The SparkSession and its cluster configuration become a compute context or a cluster context, and there is no session to keep alive.
  • spark.read.parquet on a directory becomes pl.scan_parquet with a glob, or with hive_partitioning=True for key=value layouts. Nothing runs until the query is submitted.
  • write.parquet becomes .remote(ctx).sink_parquet(...).await_result(), and .remote(ctx).show() previews rows. See Run a remote query.
  • Column expressions combine differently from Spark's row-based DataFrame. The Coming from Apache Spark page of the Polars user guide covers those API differences.

What does not move

  • Continuous stream processing: Polars Nanook runs a bounded query and stops.
  • A SQL endpoint for many concurrent users: results are written to storage for other tools to read.
  • Hosted notebooks or experiment tracking: not part of Polars Nanook.
  • Python UDFs and expression plugins do move, provided their packages are available on the compute. See Use Python packages and plugins.

Size the compute from the memory the job needs, measured with the profiler, not from the old cluster size (Scale a query), and run it from Airflow, Dagster, Prefect or Lambda with a service account.