1BRC in Python with Dask and Spark

[scharlottej13]

Inspired by #62, wanted to share a Dask and PySpark implementation for the challenge. Both were run on an M1 Mac (8-core CPU, 16 GB memory). Dask took 32.8 seconds and Spark took 2min 1s.

Worth noting the Dask implementation uses Dask Expressions, which is under active development. Dask users can expect to see these changes in core Dask DataFrame soon.

More details in our blog post.

Dask implementation - 32.8 seconds (± 2.43 s across 7 trials)

from dask.distributed import Client
import dask_expr as dd

client = Client()

df = dd.read_csv(
    "measurements.txt",
    sep=";",
    header=None,
    names=["station", "measure"],
    engine='pyarrow',
    dtype_backend='pyarrow'
)
df = df.groupby("station").agg(["min", "max", "mean"])
df.columns = df.columns.droplevel()
df = df.sort_values("station").compute()

Spark implementation - 2min 1s (± 6.45 s across 7 trials)

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, FloatType, StringType

spark = SparkSession.builder.master("local[*]").getOrCreate()
schema = StructType(
    [
        StructField("station", StringType(), True),
        StructField("measure", FloatType(), True),
    ]
)

df = spark.read.options(delimiter=";", header=False).csv(
    "repos/1brc/measurements.txt", schema=schema
)
df.createOrReplaceTempView("measurements")

query = """
SELECT station, MAX(measure) as max, MIN(measure) as min, AVG(measure) as mean
FROM measurements
GROUP BY station ORDER BY station;
"""

result = spark.sql(query).show()