Replies: 2 comments 4 replies
|
Already has flink support in the main repo under |
|
Yeah @zhanglistar gluten-flink is doing great work — I've been following the checkpoint PRs. The case I'm curious about is a bit different: teams already running Spark Structured Streaming who aren't going to rewrite in Flink. Those queries all fall back to vanilla Spark under Gluten today, and I've been trying to work out whether that's a deliberate call or just something nobody's gotten to. What made me look again is Real-Time Mode landing in Apache Spark 4.1 (SPIP SPARK-52330) it's upstream now. Reading through it, RTM is stateless-only and record-at-a-time, so it's honestly not somewhere Velox would help. But it does mean upstream is taking the low-latency end of streaming and leaving the throughput end open and the throughput end is the part Gluten is good at. One thing I noticed in the 4.1 source that may matter to Gluten regardless of any of the above: Do you know if native Spark streaming has come up before? cc: @zhouyuan @FelixYBW @philo-he Any thoughts? On supporting this ? |
|
@malinjawi No. We have never had any customer show interest in Spark streaming support in Gluten. More companies are interested in Flink. |
|
Hey @FelixYBW thanks for weighing in. This was purely an engineering play not requested by any customer. Currently, Databricks is doing exactly this with Photon extending it to Structured Streaming, including stateful operators (Data+AI Summit session). The demand may exist in the future but it wont be reaching Gluten as a feature today. Either way, I'll repro the Spark 4.1 RealTimeModeAllowlist issue and file it separately, since it affects Gluten on 4.1 regardless. |
|
@malinjawi, to support this, do we also need to make multiple code changes in Velox? |
@malinjawi Thanks for your information. I think that databricks people extend photon to support support structed streaming computation is right and obvious since they want customers use spark to meet all data processing need. But, I think that in this point and the near future, Flink is still widely used in the world. So supporting flink in Guten is maybe right. And, welcome to contribute to RFCs and PRs to support spark structured streaming. |
Uh oh!
There was an error while loading. Please reload this page.
What happens when Structured Streaming stops falling back — spike findings on the Velox backend
Spark Structured Streaming is Gluten's blind spot. There are no native streaming operators, no streaming-aware validation, and every stateful operator in a streaming query runs on vanilla Spark — the community's streaming work so far has mostly been about making fallback safe. I couldn't find anyone who had measured whether it has to stay that way, so I ran a spike to find out what happens when streaming plans go native, including the stateful part.
The result is a POC branch and numbers, some flattering, some not. It's a research snapshot —default-off behind gates, rough in places — not proposed for merge as-is.
POC branch: https://github.com/malinjawi/incubator-gluten/tree/velox-native-streaming-poc
What's in the spike
StateStoreProviderAPI, with delta checkpointing: O(changed) per commit, periodic snapshots, replay on restart.The numbers (one machine, so directional)
Everything below survived a 5-repeat, warmup-dropped, ±2-sigma protocol; single-shot "wins" that failed it were discarded.
Stateless, compute-bound work wins at scale. Parquet micro-batches joined to a broadcast dim table with ~29 derived columns: native runs 1.5-1.8x vanilla at 10M rows/batch (conservative bound ~1.5x, mean ~1.8x), bit-exact parity. The win shrinks with batch size and inverts below ~1M rows/batch — Velox's vectorized kernels beat codegen on heavy expression work, but a fixed per-trigger cost only amortizes when batches are large. "Does native streaming help?" has no scalar answer; it's a batch-size curve, and your batch size tells you which side you're on.
Stateful started at 0.32x, for three measurable reasons. My first stateful implementation lost badly at 1M. Instrumentation traced the loss to: (1) the row/columnar bridges around the native island (the streaming scan was left vanilla); (2) per-key JNI marshaling with UnsafeRow encode/decode — about 83% of per-batch cost — eliminated by a typed resident map plus bulk columnar I/O; (3) an O(total-state) full snapshot on every commit, replaced with delta checkpointing, after which commit time went flat. The fixed implementation reaches 0.95x — parity — at 1M. At-scale throughput of the clean stateful path is unmeasured; I'm publishing parity as parity, not extrapolating a win.
There's a hard ceiling worth naming. While Spark owns the control plane (triggers, offset WAL, checkpoint commits, sink epochs), there's a per-trigger floor of ~100-200ms in my runs, identical on both engines, so Amdahl caps end-to-end speedup around 2.2x no matter how fast the native island gets.
One side finding. The typed native state layout held the same 10M-key state in roughly 5-10x less memory than the vanilla provider's maps (~100-300MB vs ~0.8-2.4GB), with the caveat that the two self-report through different mechanisms. Even at throughput parity, state density might matter on its own for large-state jobs.
The architecture bet, and where it touches the Flink lane
The shape that survived: the JVM keeps the entire state lifecycle — open, commit, checkpoint, maintenance, metrics — while native code executes only the per-batch work and hands changelog bytes back to be written through Spark's own checkpoint machinery. A fully native state store looked attractive and I rejected it: it creates two coexisting state-management systems whose persistent formats must stay switchable between native and vanilla execution.
The part I'd most like gluten-flink eyes on (#12164, #12459): the C++ state kernels are deliberately byte-oriented — no Spark types, no JNI types — and the operator output contract reserves a retraction/RowKind channel, because Spark micro-batch and Flink differ in the checkpoint adapter (versioned store vs barrier snapshots), not in the state/op kernels. In principle that layer could be called from velox4j as easily as from Gluten's Spark JNI. Genuine question, not a pitch: is a shared layer below the checkpoint contract worth pursuing together, or do the Flink lane's state-backend plans point somewhere else?
Limitations, plainly
Spark 4.1-only. Narrow operator set: no stream-stream joins, session windows, or transformWithState. Stateful runs single-partition — the streaming shuffle isn't gated for native yet. A known process-exit-only teardown crash on the native file-source scan path is worked around via the data cache. Every number above is one laptop, one partition. If the evidence doesn't survive broader validation, the right outcome is that the gates stay off.
What I'm asking
Happy to share the measurement harness or expand any section into a design write-up. If the discussion lands on "not worth it," writing that down with data attached is a valid outcome of a spike too — the next person who wonders won't have to build one to find out.
All reactions