[Feature Request]: A dedicated Spark pipeline execution engine

[mattcasters]

What would you like to happen?

While the Beam implementation has a Spark runner and we do have a BeamSparkPipelineEngine, it's running a few years behind in version support for Apache Spark. The Apache Beam API is slow to adopt the new versions.
Beyond that it's likely interesting to offer support for spark-specific transforms for SQL and MLib functionality. The downside would be the lack of windowing and streaming support but we'd have a fallback to Beam for that.
The Spark version we should support is 4.1.x
For Beam we have a mini single-threaded pipeline feeding data into a single transform. We should investigate more options to reduce the overhead there.

Just some considerations to make the transforms run in Spark...

Implement MapPartitionsFunction

Wrap the Hop transform inside a class that implements Spark’s MapPartitionsFunction<Row, Row>. This ensures the initialization overhead of the Hop engine only happens once per data partition rather than for every single row.

Apply it to the Spark Pipeline

Once your wrapper is built, you can easily plug any Hop transformation metadata directly into your standard Spark Java code using the .mapPartitions() API

Things to look out for

• Serialization (The NotSerializableException): Hop metadata objects (TransformMeta, PipelineMeta) are heavily object-oriented and often contain references that cannot be natively serialized across network boundaries. You must serialize them into an XML string or JSON text on the Driver node, pass that string to the wrapper constructor, and fully deserialize/reconstruct the Hop objects inside the call() method on the Executor.
• Schema Management: Hop relies heavily on IRowMeta to dynamically define column positions and data types at runtime. Spark requires a strictly defined structural schema (StructType) before compilation. Your interface must dynamically map Spark’s StructType fields to Hop’s IRowMeta array elements and back.
• Stateful Hop Transforms: Standard stateless transforms (like String operations or Calculator steps) will work perfectly in this setup. However, stateful Hop transforms that rely on sorting (Sort Rows), windowing, or full-dataset aggregation (Group By) will fail or give wrong answers. This happens because they will only sort/aggregate data within that single Spark partition instead of performing a cluster-wide Spark Shuffle. We need to either block usage (Sort Rows) or convert functionality to Spark variants (Group By)

Issue Priority

Priority: 3

Issue Component

Component: Pipelines