spark-mf4 (v2) — asammdf-backed Spark reader

v2 changes the approach. Instead of a hand-written MDF parser (v1, the Scala data source), v2 delegates parsing to the mature asammdf library and distributes the work across Spark. This reads the real-world files (CANedge, Vector, ETAS, vendor variants) that a hand-written parser can silently fail on — e.g. returning 0 rows — because asammdf handles the full spec.

Install (Databricks)

pyspark and pyarrow are already on the cluster. Install the library:

%pip install spark-mf4
dbutils.library.restartPython()

Use

from spark_mf4 import read_mf4

df = read_mf4(spark, "/mnt/reservoir/preet/0000002.MF4")   # a file, dir, or glob
df.display()

Paths — use a Spark path, not a local one

read_mf4 reads through Spark, so give it a Spark filesystem path:

The Python/FUSE form /dbfs/mnt/…​ is auto-rewritten to /mnt/…​, so both work. Extensions are matched case-insensitively (.MF4, .mf4, .Mdf, …).

How it works — and how it uses Spark parallelism

 driver: list files                 executors (in parallel, one task per file)
 ┌───────────────────────┐          ┌───────────────────────────────────────────┐
 │ spark.read            │  files   │ mapInPandas(partition):                    │
 │   .format("binaryFile")│ ───────► │   for (path, content) in partition:        │
 │   .load(path)         │  as N    │     MDF(BytesIO(content))  ← asammdf        │
 │  → (path, content)    │  tasks   │       .to_dataframe() → pandas → Arrow rows │
 └───────────────────────┘          └───────────────────────────────────────────┘
                                       shuffle-free; results union into one DataFrame

Step by step:

1. List + read files (binaryFile). Spark's built-in binaryFile source enumerates every matching file under path on whatever storage Spark supports (mounts, Volumes, DBFS, S3/ADLS). It yields a row per file: (path, content) where content is the file's raw bytes. This is a normal Spark scan, so file listing and reading happen on the cluster, not the driver.

2. One task per file (the unit of parallelism). binaryFile does not split a file — each file becomes one input partition, and Spark schedules those partitions across all executor cores. So N files are decoded concurrently, up to the number of cores in the cluster. 200 files on a 40-core cluster → ~40 decoded at a time, fully utilised.

3. Decode on the executor (mapInPandas + asammdf). For each file in its partition, the executor hands the in-memory bytes to asammdf.MDF(BytesIO(content)) and calls to_dataframe(), producing a pandas frame that PySpark converts to Arrow batches — the rows of the output DataFrame. This runs inside the Python worker on the executor, in parallel with every other task. There is no shuffle — it's a map-only stage — so it scales linearly with cores until I/O-bound.

4. Union. The per-file DataFrames are the partitions of the returned Spark DataFrame; downstream operations (filter/agg/write) parallelise normally.

What this means (and the limits — read this)

• ✅ Scales with the number of files. Many small/medium files → excellent parallelism, no driver bottleneck, no shuffle.
• ⚠️ No intra-file parallelism. A single file is decoded by exactly one task on one core (asammdf is not distributed). One huge file → single-core decode; the fix is to log/split into more files, or accept single-file speed.
• ⚠️ Per-file memory. binaryFile loads a file's entire bytes into the executor, and asammdf materialises the frame in memory, so an executor needs roughly file_size + decoded_frame_size of RAM per concurrent task. Size executors accordingly; use raster= to down-sample very high-rate data.
• ⚠️ binaryFile size cap. Files larger than spark.sql.sources.binaryFile.maxLength (default ~2 GiB) error; raise that conf if you have bigger files: spark.conf.set("spark.sql.sources.binaryFile.maxLength", 8*1024*1024*1024).
• ℹ️ Schema is inferred from the first file; read files of the same type together (heterogeneous files share the first file's columns).

Options

CAN/LIN signal decoding (DBC)

Point at raw CAN logs and a DBC database, and the reader decodes each frame into physical signals (via asammdf extract_bus_logging) — the output columns become the DBC signal names:

df = read_mf4(spark, "/mnt/reservoir/preet/", dbc="/mnt/reservoir/dbc/signals.dbc")
df.select("timestamp", "EngineSpeed", "VehicleSpeed").display()

The DBC is read once on the driver and its bytes are shipped to the executors, so decoding runs per file, in parallel, exactly like a normal read. Files with no CAN bus-logging groups yield no signal rows (they're skipped gracefully).

Notes

• v1 (the pure-Scala format("mf4") data source) remains on main; v2 lives here under python/.