Metadata-Version: 2.4
Name: spark-mf4
Version: 2.1.0
Summary: Read ASAM MDF/MF4 measurement files into Apache Spark, backed by asammdf
Author-email: Preet Ranjan <preetish.888@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/PreetRanjan/spark-mf4
Keywords: asam,mdf,mf4,spark,databricks,can,automotive
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: asammdf>=8.0
Requires-Dist: pandas>=1.5
Requires-Dist: numpy
Provides-Extra: spark
Requires-Dist: pyspark>=3.4; extra == "spark"
Requires-Dist: pyarrow; extra == "spark"

# 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`](https://github.com/danielhrisca/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:

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

## Use

```python
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**:

| Storage | Path to use |
|---------|-------------|
| ADLS/S3 **mounted** | `/mnt/<mount>/…​` (or `dbfs:/mnt/<mount>/…​`) |
| Unity Catalog Volume | `/Volumes/<cat>/<schema>/<vol>/…​` |
| DBFS | `dbfs:/…​` or `/…​` |
| Cloud direct | `abfss://…​`, `s3://…​`, `gs://…​` |

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

| Option | Default | Meaning |
|--------|---------|---------|
| `channels` | `None` | Restrict to these channel names |
| `channel_group` | `None` | Read only this channel group (native time base) instead of the merged wide table |
| `raster` | `None` | Resample to this many seconds (also reduces memory) |
| `time_from_zero` | `False` | Shift timestamps to start at 0 |
| `add_source_column` | `True` | Append the source file path as `source_file` |
| `extensions` | `("mf4","mdf","dat")` | Case-insensitive file extensions to include (`None` = any) |
| `path_glob` | `None` | Spark `pathGlobFilter` (overrides `extensions`); use `"*"` to read any file |
| `recursive` | `True` | Recurse into sub-directories |
| `dbc` | `None` | Spark path to a `.dbc`; decode raw CAN frames into physical signals |

## 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:

```python
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/`.
