Transform Executor: Multi-Engine Adapter Layer
Tags: #TransformExecutor #MultiEngine #Flink #Spark #DuckDB #coomia-dip
“Series: S3 Data Foundation · Article 20 | Level: Advanced | Reading Time: 20 min
Transform Executor: Multi-Engine Adapter Layer
Tags: #TransformExecutor #MultiEngine #Flink #Spark #DuckDB #coomia-dip
#TL;DR
Transform steps in Pipeline DSL ultimately execute on specific compute engines. The coomia-dip Transform Executor is a multi-engine adapter layer — the same Transform definition compiles to Flink DataStream operators, Spark DataFrame operations, or DuckDB SQL statements. This article fully dissects the Transform Executor's architecture, including the unified Transform interface, Flink/Spark/DuckDB adapter implementations, automatic engine selection strategy, cross-engine schema consistency guarantees, performance benchmark comparisons, and failover mechanisms.
#1. Why Multi-Engine Adaptation
#1.1 Engine Strengths by Scenario
Three-Engine Comparison:
+-----------------+--------------+--------------+--------------+
| Scenario | Flink | Spark | DuckDB |
+-----------------+--------------+--------------+--------------+
| Real-time stream | 5/5 | 3/5 | 1/5 |
| CDC ingestion | 5/5 | 2/5 | 1/5 |
| Large batch | 3/5 | 5/5 | 2/5 |
| Interactive | 2/5 | 3/5 | 5/5 |
| Small dataset | 2/5 | 2/5 | 5/5 |
| Window agg | 5/5 | 4/5 | 2/5 |
| Embedded exec | No | No | 5/5 |
| Resource usage | High | High | Very Low |
+-----------------+--------------+--------------+--------------+
#1.2 Value of a Unified Interface
Unified Transform Interface Value:
User perspective:
pipeline.transform(filter(lambda r: r["age"] > 30))
-> Don't care if it's Flink/Spark/DuckDB underneath
System perspective:
-> Auto-select engine based on data volume, latency, resources
-> Auto-failover when one engine is down
-> Dev/test uses DuckDB (instant start), production uses Flink
#2. Transform Interface Definition
#2.1 Unified Interface
from abc import ABC, abstractmethod
import pyarrow as pa
class TransformOperator(ABC):
"""Unified Transform operator interface"""
@abstractmethod
def name(self) -> str: ...
@abstractmethod
def input_schema(self) -> pa.Schema: ...
@abstractmethod
def output_schema(self) -> pa.Schema: ...
@abstractmethod
def compile_flink(self) -> 'FlinkOperator': ...
@abstractmethod
def compile_spark(self) -> 'SparkTransform': ...
@abstractmethod
def compile_duckdb(self) -> str: ...
class FilterOperator(TransformOperator):
def __init__(self, predicate: str | Callable):
self._predicate = predicate
def output_schema(self) -> pa.Schema:
return self.input_schema()
def compile_flink(self) -> 'FlinkFilterFunction':
if callable(self._predicate):
return FlinkLambdaFilter(self._predicate)
return FlinkExprFilter(self._predicate)
def compile_spark(self) -> Column:
if callable(self._predicate):
return SparkUDFFilter(self._predicate)
return F.expr(self._predicate)
def compile_duckdb(self) -> str:
if callable(self._predicate):
raise CompilationError(
"Lambda predicates not supported in DuckDB mode"
)
return f"WHERE {self._predicate}"
class AggregateOperator(TransformOperator):
def __init__(self, group_by: list[str], aggregations: dict):
self._group_by = group_by
self._aggregations = aggregations
def compile_flink(self) -> 'FlinkAggregation':
return FlinkKeyedAggregation(
key_selector=self._group_by,
aggregations=self._aggregations
)
def compile_spark(self) -> 'SparkGroupedAgg':
return SparkGroupByAgg(
group_cols=self._group_by,
agg_exprs=self._aggregations
)
def compile_duckdb(self) -> str:
agg_parts = []
for alias, (col, func) in self._aggregations.items():
agg_parts.append(f"{func.upper()}({col}) AS {alias}")
group_cols = ', '.join(self._group_by)
return (
f"SELECT {group_cols}, {', '.join(agg_parts)} "
f"GROUP BY {group_cols}"
)
#3. Flink Adapter
class FlinkTransformCompiler:
"""Compile Transform chain to Flink DataStream operations"""
def compile(
self, transforms: list[TransformOperator],
source_stream: DataStream
) -> DataStream:
current = source_stream
for transform in transforms:
flink_op = transform.compile_flink()
if isinstance(flink_op, FlinkFilterFunction):
current = current.filter(flink_op)
elif isinstance(flink_op, FlinkMapFunction):
current = current.map(
flink_op,
output_type=self._to_flink_type(
transform.output_schema()
)
)
elif isinstance(flink_op, FlinkKeyedAggregation):
current = (
current
.key_by(flink_op.key_selector)
.window(flink_op.window_assigner)
.aggregate(flink_op.aggregate_function)
)
return current
#4. DuckDB Adapter
class DuckDBTransformCompiler:
"""Compile Transform chain to DuckDB SQL"""
def compile(
self, transforms: list[TransformOperator],
source_sql: str
) -> str:
current_sql = f"({source_sql}) AS source"
for i, transform in enumerate(transforms):
sql_fragment = transform.compile_duckdb()
alias = f"t{i}"
if sql_fragment.startswith("WHERE"):
current_sql = (
f"(SELECT * FROM {current_sql} "
f"{sql_fragment}) AS {alias}"
)
elif sql_fragment.startswith("SELECT"):
current_sql = (
f"({sql_fragment} FROM {current_sql}) AS {alias}"
)
return f"SELECT * FROM {current_sql}"
#5. Automatic Engine Selection
class EngineSelector:
"""Automatic Transform execution engine selection"""
def select(self, pipeline: Pipeline) -> Engine:
features = self._analyze_pipeline(pipeline)
if features.has_cdc_source:
return Engine.FLINK
if features.has_window_aggregation:
return Engine.FLINK
if features.estimated_data_size_mb < 100:
return Engine.DUCKDB
if features.estimated_data_size_mb > 10_000:
return Engine.SPARK
return Engine.DUCKDB
#6. Failover Mechanism
class EngineFailover:
"""Engine failover handler"""
async def execute_with_failover(
self, pipeline: Pipeline, primary_engine: Engine
) -> ExecutionResult:
fallback_order = self._get_fallback_order(primary_engine)
for engine in [primary_engine] + fallback_order:
try:
return await self._execute_on(pipeline, engine)
except EngineUnavailableError as e:
logger.warning(
f"Engine {engine} unavailable: {e}. Trying next..."
)
except CompilationError as e:
logger.warning(
f"Cannot compile for {engine}: {e}. Trying next..."
)
raise AllEnginesFailedError(
f"All engines failed for '{pipeline.name}'"
)
def _get_fallback_order(self, primary: Engine) -> list[Engine]:
return {
Engine.FLINK: [Engine.SPARK, Engine.DUCKDB],
Engine.SPARK: [Engine.FLINK, Engine.DUCKDB],
Engine.DUCKDB: [Engine.FLINK, Engine.SPARK],
}[primary]
#7. Performance Benchmarks
Three-Engine Performance Comparison:
Scenario: 1M rows, filter -> map -> aggregate
+------------------+----------+----------+----------+
| Metric | Flink | Spark | DuckDB |
+------------------+----------+----------+----------+
| Startup time | 15s | 30s | 0.1s |
| Processing time | 8s | 12s | 3s |
| Total time | 23s | 42s | 3.1s |
| Memory usage | 2GB | 4GB | 200MB |
| Throughput (row/s)| 125K | 83K | 333K |
+------------------+----------+----------+----------+
| Stream latency P99| 100ms | 2s | N/A |
| Window agg | Native | Micro | No |
| Fault recovery | Seconds | Minutes | None |
+------------------+----------+----------+----------+
Conclusions:
- Small batch -> DuckDB wins (7-14x faster)
- Streaming -> Flink wins (native stream engine)
- Large batch -> Spark wins (distributed scale)
#8. Testing Strategy
class TestTransformExecutor:
def test_filter_compiles_to_all_engines(self):
op = FilterOperator("age > 30")
assert op.compile_flink() is not None
assert op.compile_spark() is not None
assert "WHERE age > 30" in op.compile_duckdb()
def test_aggregate_consistency_across_engines(self):
"""All three engines should produce identical results"""
data = generate_test_data(1000)
op = AggregateOperator(
group_by=["type"],
aggregations={"total": ("value", "sum")}
)
flink_result = execute_flink(op, data)
spark_result = execute_spark(op, data)
duckdb_result = execute_duckdb(op, data)
assert_results_equal(flink_result, spark_result)
assert_results_equal(spark_result, duckdb_result)
def test_engine_auto_selection(self):
small = Pipeline.create("s").source(csv_file("10mb.csv"))...
assert EngineSelector().select(small) == Engine.DUCKDB
cdc = Pipeline.create("c").source(mysql_cdc("db.t"))...
assert EngineSelector().select(cdc) == Engine.FLINK
#Key Takeaways
-
The unified Transform interface decouples users from engines: the same filter/map/aggregate definition compiles to Flink, Spark, or DuckDB — users never need to know the underlying engine.
-
Automatic engine selection is based on data characteristics: CDC source -> Flink, small datasets -> DuckDB, large batch -> Spark — the system decides based on data volume and processing needs.
-
DuckDB dominates small-data scenarios in performance: instant startup (0.1s vs 15-30s), fast single-threaded processing, making it the best choice for dev/test.
-
Failover ensures production reliability: when the primary engine is unavailable, automatic switching to backup engines prevents Pipeline interruption.
-
Cross-engine result consistency is the core constraint: all three engines must produce identical results for the same data and Transform, guaranteed through the unified interface and consistency tests.
#Next Article
Next up: S3-21 "World Transform: Global Data Consistency Transformation" will introduce the Palantir Foundry-style World Transform concept — how to perform atomic global transformations on the entire Ontology world state.
Tags: #TransformExecutor #MultiEngine #Flink #Spark #DuckDB #EngineSelection #Failover #Consistency #coomia-dip #DataFoundation