Streamline is a novel dynamic multi-layer auto-tuning framework for data pipeline ensembles that optimizes throughput, end-to-end latency, and resource utilization while reducing monetary cost. It delivers a holistic auto-tuning and scheduling approach to ensure efficient processing across an ensemble of data pipelines.
# Clone repository
git clone https://github.com/UIBK-DPS-DC/Streamline.git
cd Streamline
# Create and activate virtual environment
python3 -m venv env
source env/bin/activate
# Install dependencies
pip install -r requirements.txtTip
We strongly recommend checking the provided examples to enhance understanding of Streamline.
For sample code and implementation guidance, refer to the examples/ directory. It contains use cases that show how to integrate and use Streamline.
# Run an example of a single data pipeline
python3.8 examples/single_pipeline.py
# Run an example of an ensembles of data pipelines
python3.8 examples/ensemble.pyStreamline starts by training the models for the transformer, neural network, and polynomial regression. Note that this step happens offline and is only needed if they haven't been trained yet.
streamline.training_manager - INFO: Starting Workload Predictor training using Transformers...
streamline.workload_predictor - INFO: Epoch 0/200, Loss: 1.2220
streamline.workload_predictor - INFO: Epoch 100/200, Loss: 0.0130
streamline.workload_predictor - INFO: Epoch 200/200, Loss: 0.0052
streamline.training_manager - INFO: Starting Neural Network training...
streamline.neural_network - INFO: Epoch 0/200, Train Loss: 0.0946, Val Loss: 0.0582
streamline.neural_network - INFO: Epoch 100/200, Train Loss: 0.0035, Val Loss: 0.0065
streamline.neural_network - INFO: Epoch 200/200, Train Loss: 0.0027, Val Loss: 0.0075
streamline.neural_network - INFO: R^2 Score: Train=0.96, Val=0.90
streamline.training_manager - INFO: Starting Polynomial Regression training...Once a new data pipeline is started, Streamline proceeds to predict the workload of operators using the pre-trained transformer model and the aspect ratio of operators.
streamline.controller - INFO: Predicting workload of operators using Transformers and aspect ratio...An example prediction for the source operator v_0 would be:
Based on the predicted load, NSGA-II is used to determine the configuration parameters for each operator.
streamline.controller - INFO: Best configuration for v_0: {'parallelism': 4.0, 'segment-size': 16384.0, 'est_tp': 1716.0094958496093, 'est_l': 178.6549966430664, 'est_cpu': 1.83804749250412}.
streamline.controller - INFO: Best configuration for v_1: {'parallelism': 4.0, 'segment-size': 16384.0, 'est_tp': 858.0047479248046, 'est_l': 97.51335861206054, 'est_cpu': 1.4166296708583832}.
streamline.controller - INFO: Best configuration for v_2: {'parallelism': 4.0, 'segment-size': 16384.0, 'est_tp': 858.0047479248046, 'est_l': 94.8901594543457, 'est_cpu': 1.3292160773277282}.
streamline.controller - INFO: Best configuration for v_3: {'parallelism': 4.0, 'segment-size': 16384.0, 'est_tp': 2315.8579250009902, 'est_l': 180.39547729492188, 'est_cpu': 3.004585027694702}.The operator instances are then scheduled on resources.
streamline.scheduler - INFO: Usable CPU time of already used resources: 0
streamline.scheduler - INFO: Total required CPU: 7.588478268384933
{
'R1': ['pipeline_1_v_0_1', 'pipeline_1_v_0_3', 'pipeline_1_v_1_1', 'pipeline_1_v_1_3', 'pipeline_1_v_2_1', 'pipeline_1_v_2_3'],
'R11': ['pipeline_1_v_3_0', 'pipeline_1_v_3_1', 'pipeline_1_v_3_2', 'pipeline_1_v_3_3', 'pipeline_1_v_0_0', 'pipeline_1_v_0_2', 'pipeline_1_v_1_0', 'pipeline_1_v_1_2', 'pipeline_1_v_2_0', 'pipeline_1_v_2_2']
}Once another data pipeline is started, the same steps are followed. First, the future workload is predicted and suitable configuration parameters are determined.
streamline.controller - INFO: Predicting workload of operators using Transformers and aspect ratio...
streamline.controller - INFO: Best configuration for v_5: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 916.0249955606461, 'est_l': 138.36511642456054, 'est_cpu': 1.3969868558645249}.
streamline.controller - INFO: Best configuration for v_6: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 916.0249955606461, 'est_l': 105.27256454467773, 'est_cpu': 1.3420857819914818}.
streamline.controller - INFO: Best configuration for v_7: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 2002.682036063938, 'est_l': 181.96080520629883, 'est_cpu': 2.9487196230888366}.Then, operators are scheduled. Streamline efficiently considers resources utilized by prior data pipelines.
streamline.scheduler - INFO: Usable CPU time of already used resources: 2.011521731615066
streamline.scheduler - INFO: Total required CPU: 5.687792260944844
{
'R11': ['pipeline_2_v_6_4'],
'R12': ['pipeline_2_v_7_0', 'pipeline_2_v_7_1', 'pipeline_2_v_7_2', 'pipeline_2_v_7_3', 'pipeline_2_v_7_4', 'pipeline_2_v_5_0', 'pipeline_2_v_5_1', 'pipeline_2_v_5_2', 'pipeline_2_v_5_3', 'pipeline_2_v_5_4', 'pipeline_2_v_6_0', 'pipeline_2_v_6_1', 'pipeline_2_v_6_2', 'pipeline_2_v_6_3']
}Streamline can be scheduled to run at regular intervals, automatically verifying whether configuration parameter or scheduling updates are required. In this case, both data pipelines are re-evaluated: no updates are necessary for the second data pipeline, while the first data pipeline is updated.
streamline.controller - INFO: Predicting workload of operators using Transformers and aspect ratio...
streamline.controller - INFO: Best configuration for v_0: {'parallelism': 2.0, 'segment-size': 16384.0, 'est_tp': 279.5950734175739, 'est_l': 92.20072036743164, 'est_cpu': 0.46633761912584304}.
streamline.controller - INFO: Best configuration for v_1: {'parallelism': 2.0, 'segment-size': 16384.0, 'est_tp': 141.17301223754882, 'est_l': 75.95237106323242, 'est_cpu': 0.3235232400894165}.
streamline.controller - INFO: Best configuration for v_2: {'parallelism': 2.0, 'segment-size': 16384.0, 'est_tp': 141.17301223754882, 'est_l': 75.88439895629882, 'est_cpu': 0.2880781376361847}.
streamline.controller - INFO: Best configuration for v_3: {'parallelism': 4.0, 'segment-size': 16384.0, 'est_tp': 938.5496359453845, 'est_l': 139.41199645996093, 'est_cpu': 1.4749153592437505}.
streamline.controller - INFO: Predicting workload of operators using Transformers and aspect ratio...
streamline.controller - INFO: Best configuration for v_5: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 919.3685557937622, 'est_l': 138.29559829711914, 'est_cpu': 1.4028689235448837}.
streamline.controller - INFO: Best configuration for v_6: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 919.3685557937622, 'est_l': 105.82319854736328, 'est_cpu': 1.3324667251110076}.
streamline.controller - INFO: Best configuration for v_7: {'parallelism': 5.0, 'segment-size': 16384.0, 'est_tp': 2020.5297649170125, 'est_l': 183.59389602661133, 'est_cpu': 3.0061577421426775}.
streamline.scheduler - INFO: Usable CPU time of already used resources: 7.1122077390551555
streamline.scheduler - INFO: Total required CPU: 2.5528543560951946
{
'R11': ['pipeline_1_v_3_0', 'pipeline_1_v_3_1', 'pipeline_1_v_3_2', 'pipeline_1_v_3_3', 'pipeline_1_v_0_0', 'pipeline_1_v_0_1', 'pipeline_1_v_1_0', 'pipeline_1_v_1_1', 'pipeline_1_v_2_0', 'pipeline_1_v_2_1']
}Resulting in a total of 2 resources used and the following CPU utilization. We note that although R1 was initially used, the periodic invocation led to its removal.
Resources Utilization:
R11: 0.35265893906168644
R12: 0.6774218880683186If the alpha threshold would be high enough (in this case, approximately 0.52, calculated as > (0.35 + 0.67) / 2), then all operators across data pipelines would be rescheduled to achieve higher
average CPU utilization and lower monetary cost.
streamline.controller - INFO: Rescheduling as average CPU utilization 0.5150404135650025 is below alpha 0.52
streamline.scheduler - INFO: Usable CPU time of already used resources: 0
streamline.scheduler - INFO: Total required CPU: 8.24064661704004
{
'R1': ['pipeline_1_v_3_1', 'pipeline_1_v_3_3', 'pipeline_2_v_5_1', 'pipeline_2_v_5_3', 'pipeline_2_v_6_0', 'pipeline_2_v_6_2', 'pipeline_2_v_6_4', 'pipeline_1_v_1_0', 'pipeline_1_v_1_1', 'pipeline_1_v_2_1'],
'R11': ['pipeline_2_v_7_0', 'pipeline_2_v_7_1', 'pipeline_2_v_7_2', 'pipeline_2_v_7_3', 'pipeline_2_v_7_4', 'pipeline_1_v_3_0', 'pipeline_1_v_3_2', 'pipeline_2_v_5_0', 'pipeline_2_v_5_2', 'pipeline_2_v_5_4', 'pipeline_2_v_6_1', 'pipeline_2_v_6_3', 'pipeline_1_v_0_0', 'pipeline_1_v_0_1', 'pipeline_1_v_2_0']
}
Resources Utilization:
R1: 0.6422665500175209
R11: 0.7089475521212445This project is licensed under the GPLv3 - see the LICENSE file for details.
Stefan Pedratscher, Zahra Najafabadi Samani, Juan Aznar Poveda, Thomas Fahringer, Marlon Etheredge, Abolfazl Younesi, Juan Jose Durillo Barrionuevo, Peter Thoman, STREAMLINE: Dynamic and Resource-Efficient Auto-Tuning of Stream Processing Data Pipeline Ensembles, Internet of Things, Volume 34, 2025, 101731, ISSN 2542-6605, https://doi.org/10.1016/j.iot.2025.101731.