|
| 1 | +<!-- |
| 2 | +Licensed to the Apache Software Foundation (ASF) under one |
| 3 | +or more contributor license agreements. See the NOTICE file |
| 4 | +distributed with this work for additional information |
| 5 | +regarding copyright ownership. The ASF licenses this file |
| 6 | +to you under the Apache License, Version 2.0 (the |
| 7 | +"License"); you may not use this file except in compliance |
| 8 | +with the License. You may obtain a copy of the License at |
| 9 | +
|
| 10 | + http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | +
|
| 12 | +Unless required by applicable law or agreed to in writing, |
| 13 | +software distributed under the License is distributed on an |
| 14 | +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | +KIND, either express or implied. See the License for the |
| 16 | +specific language governing permissions and limitations |
| 17 | +under the License. |
| 18 | +--> |
| 19 | + |
| 20 | +# DataFusion as a Spark data source (ADBC) |
| 21 | + |
| 22 | +This repository ships a Spark connector that lets a Spark job read from a |
| 23 | +DataFusion `TableProvider` as a native Spark `DataSourceV2`. Spark sees an |
| 24 | +ordinary external table; the data is produced by DataFusion in native code and |
| 25 | +handed back as Apache Arrow batches. |
| 26 | + |
| 27 | +## What this repository provides |
| 28 | + |
| 29 | +Two pieces work together: |
| 30 | + |
| 31 | +- **The `adbc-datafusion` Spark connector** (`spark/`) — a Spark |
| 32 | + `DataSourceV2` registered under the format name `adbc-datafusion`. It builds |
| 33 | + the scan, pushes projection / filters / limit down, maps DataFusion's scan |
| 34 | + partitions onto Spark input partitions, and imports each partition's Arrow |
| 35 | + data into Spark **zero-copy** as an `ArrowColumnVector`. |
| 36 | +- **An example DataFusion ADBC driver** (`examples/adbc-datafusion-driver/`) — |
| 37 | + a small native library (cdylib) that exposes a DataFusion `TableProvider` |
| 38 | + through the [ADBC](https://arrow.apache.org/adbc/) C API. It is a worked |
| 39 | + example you can copy: your own driver registers whatever provider you want and |
| 40 | + is loaded the same way. |
| 41 | + |
| 42 | +The connector is built on the standard Apache Arrow ADBC Java driver manager |
| 43 | +(`adbc-core` + `adbc-driver-jni`), so it talks to the native driver through a |
| 44 | +stable, public interface. Because that boundary is plain ADBC, the same |
| 45 | +connector can front **any** ADBC-speaking source; the DataFusion driver shown |
| 46 | +here is just one example, and you can point it at a different ADBC driver |
| 47 | +without changing the connector. |
| 48 | + |
| 49 | +```text |
| 50 | +Spark DataSourceV2 (adbc-datafusion) |
| 51 | + → arrow-adbc Java driver manager (adbc-core + adbc-driver-jni) |
| 52 | + → native ADBC driver (the DataFusion cdylib) |
| 53 | + → your DataFusion TableProvider |
| 54 | +``` |
| 55 | + |
| 56 | +## Reading from Spark |
| 57 | + |
| 58 | +Point the `driver` option at the built driver library and the `table` option at |
| 59 | +a table the provider exposes: |
| 60 | + |
| 61 | +```scala |
| 62 | +val df = spark.read |
| 63 | + .format("adbc-datafusion") |
| 64 | + .option("driver", "/abs/path/to/libadbc_datafusion_example_driver.so") |
| 65 | + .option("entrypoint", "AdbcDatafusionExampleInit") |
| 66 | + .option("table", "example") |
| 67 | + .load() |
| 68 | + |
| 69 | +df.filter("id > 1").select("name").show() |
| 70 | +``` |
| 71 | + |
| 72 | +The connector probes the schema once (on the driver) when the `DataFrame` is |
| 73 | +created, then plans the scan. `df.filter(...)` and `df.select(...)` are pushed |
| 74 | +into the DataFusion scan where possible (see below); the rest Spark evaluates |
| 75 | +itself. |
| 76 | + |
| 77 | +## Two partitioning models that must be reconciled |
| 78 | + |
| 79 | +The heart of the connector is mapping DataFusion's idea of a partitioned scan |
| 80 | +onto Spark's idea of tasks. The two engines size scans on **different |
| 81 | +principles**, and a provider author who ignores the difference will get a job |
| 82 | +that runs but performs badly. |
| 83 | + |
| 84 | +### DataFusion: parallelism-bound |
| 85 | + |
| 86 | +DataFusion partitions a scan to **saturate the cores of one machine**. Its |
| 87 | +built-in `ListingTable`, for example, packs input files into roughly |
| 88 | +`target_partitions` groups — where `target_partitions` defaults to the local |
| 89 | +core count — using a minimum-size floor and optional single-file splitting only |
| 90 | +to *reach* that count. The partition count is therefore approximately |
| 91 | + |
| 92 | +```text |
| 93 | +N ≈ target_partitions ≈ cores |
| 94 | +``` |
| 95 | + |
| 96 | +independent of how much data there is. More data means **bigger** partitions, |
| 97 | +not more of them. |
| 98 | + |
| 99 | +### Spark: one task per partition, byte-bound |
| 100 | + |
| 101 | +Spark turns **each input partition into one task**, and a native Spark file |
| 102 | +source sizes partitions by **bytes**: it targets roughly one partition per |
| 103 | +`spark.sql.files.maxPartitionBytes` (default 128 MB), cutting files into splits |
| 104 | +and bin-packing them to that size. (Spark shrinks the target below 128 MB when |
| 105 | +the data is small, so that there are at least enough partitions to keep every |
| 106 | +core busy.) |
| 107 | + |
| 108 | +So large data means **many** partitions — typically far more than the cluster |
| 109 | +has cores — which run in waves. Each task processes a bounded slice (≈128 MB), |
| 110 | +which is what keeps memory bounded and lets Spark reschedule stragglers. |
| 111 | + |
| 112 | +### How the connector joins them |
| 113 | + |
| 114 | +For a scan that only pushes projection / filter / limit (no DataFusion-side |
| 115 | +join or aggregation), the physical plan's output partitioning **is** the |
| 116 | +provider's `scan()` partitioning, and the mapping to Spark is **1:1**: |
| 117 | + |
| 118 | +```text |
| 119 | +provider.scan() output partitions |
| 120 | + → DataFusion physical plan partition count N |
| 121 | + → N ADBC partition descriptors |
| 122 | + → N Spark input partitions = N Spark tasks (scan stage) |
| 123 | +``` |
| 124 | + |
| 125 | +Spark never splits or merges a partition afterward — `df.rdd().getNumPartitions() |
| 126 | +== N`, and the cluster core count only bounds how many of the `N` run at once. |
| 127 | + |
| 128 | +**The consequence you must plan for:** suppose a provider keeps DataFusion's |
| 129 | +default of `N = target_partitions ≈ cores` partitions. On a large dataset that |
| 130 | +produces only a handful of partitions, each holding about `total_bytes / cores` |
| 131 | +of data — so Spark runs a few very large tasks. Because a task is the unit of |
| 132 | +work, those oversized tasks bring memory pressure, stragglers that hold up the |
| 133 | +whole stage, and no way for Spark to rebalance. To feed Spark well, a provider |
| 134 | +should size partitions by **bytes** — the way a Spark file source does — rather |
| 135 | +than by the `target_partitions` count. |
| 136 | + |
| 137 | +## Sizing the scan |
| 138 | + |
| 139 | +Aim for the Spark sweet spot: |
| 140 | + |
| 141 | +- **Floor:** `N` ≥ total executor cores, or cores sit idle. `~2–4× cores` gives |
| 142 | + slack for skew and stragglers. |
| 143 | +- **Per-partition size:** big enough to amortize per-task overhead — the |
| 144 | + canonical target is **128–256 MB** of data per partition (matching Spark's |
| 145 | + `spark.sql.files.maxPartitionBytes` default of 128 MB). |
| 146 | +- **Ceiling — and ADBC lowers it.** Beyond Spark's generic per-task cost, this |
| 147 | + path adds two per-partition costs: each of the `N` ADBC descriptors carries a |
| 148 | + copy of the **whole serialized physical plan**, and each task **deserializes** |
| 149 | + that plan when it reads its partition. (The driver caches a deserialized plan |
| 150 | + per connection, but that does not help here: each Spark task uses its own |
| 151 | + connection and reads a single partition, so the plan is deserialized once per |
| 152 | + task and never reused.) Tens of thousands of tiny partitions is far costlier |
| 153 | + here than for a native file source. Keep `N` in the hundreds; prefer bigger |
| 154 | + partitions over a huge count. |
| 155 | + |
| 156 | +Rule of thumb, balanced by **bytes** rather than split count: |
| 157 | + |
| 158 | +```text |
| 159 | +N ≈ clamp(total_bytes / target_bytes, floor=cores, ceiling≈hundreds) |
| 160 | +``` |
| 161 | + |
| 162 | +## Writing a provider that feeds Spark well |
| 163 | + |
| 164 | +Your provider's `scan()` receives the session, so it can read the connector's |
| 165 | +parallelism hint and then decide based on what it knows about the data: |
| 166 | + |
| 167 | +```text |
| 168 | +T = state.config().target_partitions() // connector hint ≈ desired parallelism |
| 169 | +target_bytes = ~128–256 MB // bias larger for ADBC to keep N down |
| 170 | +min_bytes = floor so partitions never get tiny // ~ tens of MB |
| 171 | +
|
| 172 | +if total_bytes known AND splittable (files / row groups / key ranges): |
| 173 | + N = clamp(ceil(total_bytes / target_bytes), 1, CEILING) // byte-bound |
| 174 | + bin-pack splits into N groups BALANCED BY BYTES // not by count |
| 175 | + do not split below min_bytes; merge small splits |
| 176 | +
|
| 177 | +elif split_count known but not bytes (shards / external partitions): |
| 178 | + N = min(split_count, CEILING) // natural partitioning |
| 179 | + if split_count >> T: coalesce shards into ~T balanced groups |
| 180 | +
|
| 181 | +else (size and splits unknown; opaque / streaming): |
| 182 | + N = T // fall back to the hint |
| 183 | + lean slightly high — stragglers hurt more than overhead |
| 184 | +``` |
| 185 | + |
| 186 | +- **Balance by bytes, not by split count.** A task is atomic; Spark cannot |
| 187 | + rebalance a fat partition. One 10 GB partition among 100 small ones stalls the |
| 188 | + whole stage. Bin-pack so partition bytes are even. |
| 189 | +- **`target_partitions` is a hint, cap, and fallback** — not the data-bound |
| 190 | + count. Use it when bytes are unknown or as a parallelism ceiling; when you know |
| 191 | + the data size, let **bytes** drive `N`. |
| 192 | + |
| 193 | +### Planning happens once, on the driver |
| 194 | + |
| 195 | +Your provider's `scan()` runs **once**, on the driver, while the multi-partition |
| 196 | +descriptors are built. The driver plans the query, fixes the partitioning, and |
| 197 | +**serializes the whole physical plan** into each descriptor. Each executor task |
| 198 | +then deserializes that plan and executes its partition index — it does **not** |
| 199 | +call `scan()` again or re-plan. So partition `index i` always means the same |
| 200 | +slice: there is no second planning pass that could disagree with the first, and |
| 201 | +no need to make `N` stable across a driver/executor boundary. |
| 202 | + |
| 203 | +The one requirement this places on a provider is that its plan be |
| 204 | +**serializable**: built-in DataFusion nodes round-trip through the default codec, |
| 205 | +but a custom `ExecutionPlan` node needs a `PhysicalExtensionCodec` registered |
| 206 | +with the driver so it can be encoded on the driver and decoded on the executor. |
| 207 | + |
| 208 | +## Connector options |
| 209 | + |
| 210 | +| Option | Required | Meaning | |
| 211 | +| --- | --- | --- | |
| 212 | +| `driver` | yes | Path to (or manifest name of) the native ADBC driver library. | |
| 213 | +| `entrypoint` | depends on driver | Name of the driver's C init symbol (e.g. `AdbcDatafusionExampleInit`). | |
| 214 | +| `table` | yes | Table the provider exposes; its schema is probed on the driver. | |
| 215 | +| `target_partitions` | no | Parallelism hint. Defaults to `SparkContext.defaultParallelism` (total executor cores). The connector issues `SET datafusion.execution.target_partitions = N` on the planning session, so it influences the partition count when the driver builds the physical plan; that plan (with its partitioning fixed) is what gets serialized into the descriptors. Pass `k × cores` to raise parallelism. | |
| 216 | +| `manifest.path` | no | Extra search path for ADBC driver manifests when `driver` is a name rather than an absolute path. | |
| 217 | +| *(anything else)* | no | Forwarded verbatim as native ADBC database options, so provider-specific knobs pass straight through. | |
| 218 | + |
| 219 | +Two things are intentionally **not** connector options: |
| 220 | + |
| 221 | +- **`maxPartitionBytes`** — byte-sizing lives in the provider's `scan()`; the |
| 222 | + connector does not know the data size and cannot derive it. Pass a sizing knob |
| 223 | + as a provider-specific option if your provider supports one. |
| 224 | +- **Reported partitioning / storage-partitioned joins |
| 225 | + (`SupportsReportPartitioning`)** — not available over ADBC today: the |
| 226 | + descriptor carries no distribution metadata, and DataFusion's hash |
| 227 | + partitioning does not match Spark's key-grouped / bucketed model. |
| 228 | + |
| 229 | +## Summary |
| 230 | + |
| 231 | +- The connector maps the provider's `scan()` partition count **1:1** onto Spark |
| 232 | + tasks. |
| 233 | +- DataFusion sizes scans to **cores** (parallelism-bound); Spark sizes them to |
| 234 | + **bytes** (one task per ~128 MB). To feed Spark well, size your provider's |
| 235 | + partitions by bytes — typically 128–256 MB each, floored at executor cores and |
| 236 | + capped in the hundreds because every descriptor carries the full serialized |
| 237 | + plan and each task deserializes it. |
| 238 | +- Planning happens once on the driver; the physical plan is serialized into each |
| 239 | + descriptor and executors run it as-is, so a custom `ExecutionPlan` node just |
| 240 | + needs a registered `PhysicalExtensionCodec` to be serializable. |
0 commit comments