LLM-driven cross-system configuration optimization for Kafka-Flink-PostgreSQL data pipelines.
Paper: PipeTune: LLM-Driven Cross-System Configuration Optimization for Multi-Component Data Pipelines (IEEE BigData 2026)
┌─────────────────────────────────────────────┐
│ PipeTune Agent │
��� │
│ Pipeline LLM Safety Config │
│ Profiler → Engine → Guardrails → Applicator│
│ ↑ │
│ Doc Store │
└──────┬────────────────────────────┬─────────┘
│ metrics │ config changes
▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────��
│ Kafka │───→│ Flink │───→│PostgreSQL│
└──────────┘ └──────────┘ └──────────┘
PipeTune reads documentation for all three systems, observes pipeline-wide metrics, and reasons about cross-system parameter interactions. A safety guardrail with atomic rollback prevents degradation during online tuning.
Prerequisites: Docker, Docker Compose, Python 3.10+, Java 11 (for Flink JAR builds)
# 1. Clone and install Python dependencies
git clone https://github.com/umur/pipetune.git
cd pipetune
pip install -r requirements.txt
# 2. Start the pipeline
docker compose up -d
# 3. Pull the LLM model
docker exec pipetune-ollama ollama pull llama3.2:3b
# 4. Run a quick smoke test (1 run per baseline/workload combo)
python run_experiments.py --quick
# 5. Run the full experiment suite (120 runs, ~2 hours)
python run_experiments.py.
├── docker-compose.yml # Single-network testbed
├── docker-compose.distributed.yml # Multi-network testbed (distributed validation)
├── Dockerfile.pipetune # PipeTune agent container
├── requirements.txt # Pinned Python dependencies
├── run_experiments.py # Main experiment runner (all 120 runs)
├── run_distributed_validation.py # Distributed validation runner
├── inject_latency.sh # tc netem latency injection for distributed setup
├── postprocess.py # Statistical analysis and table generation
├── consistency_check.py # Cross-section numerical consistency checker
├── experiments/
│ └── config.yaml # Experiment configuration (workloads, seeds, metrics)
├── src/
│ ├── pipetune/ # PipeTune agent
│ │ ├── main.py # Entry point
│ │ ├── tuner.py # Main tuning loop (Algorithm 1)
│ │ ├── llm_engine.py # LLM Reasoning Engine (Ollama/Llama 3.2 3B)
│ │ ├── profiler.py # Pipeline Profiler (Kafka JMX, Flink REST, PG stats)
│ │ ├── guardrails.py # Safety Guardrails + Rollback Engine
│ │ ├── applicator.py # Configuration Applicator (Kafka/Flink/PG)
│ │ ├── docs_store.py # Documentation Store
│ │ ├── config.py # Parameter definitions and ranges
│ │ └── docs/ # Per-system documentation (YAML)
│ ├── baselines/ # Baseline implementations
│ │ ├── default.py # B1: Factory defaults
│ │ ├── expert.py # B2: Best-practice documentation values
│ │ ├── independent_bo.py # B3: Per-component Bayesian optimization (Optuna TPE)
│ │ └── joint_bo.py # B4: Joint 20-parameter Bayesian optimization
│ ├── pipeline/ # Pipeline infrastructure
│ │ ├── event_generator.py # Synthetic workload generator
│ │ ├── flink_job_submitter.py # Flink REST API client
│ │ └── flink-job/ # Flink streaming job (Java/Maven)
│ └── experiments/ # Experiment orchestration
│ ├── runner.py # Run management
│ └── stats.py # Bootstrap CI, Mann-Whitney U, Cliff's delta
├── plots/ # Figure generation
│ ├── plot_results.py
│ ├── generate_all.py
│ └── figures/ # Generated PDF figures (committed)
├── results/ # Experiment results (committed for reproducibility)
│ ├── all_results.csv # All 120 runs
│ ├─��� distributed/ # Distributed validation results
│ ├── convergence/ # Per-iteration convergence data
│ ├── interactions.csv # Cross-system interaction isolation tests
│ ├── rollbacks.csv # Rollback validation results
│ └── statistical_tests.csv # Pairwise statistical comparisons
├── tests/ # Unit tests
└── init-scripts/ # PostgreSQL schema initialization
| ID | Name | Description | Rate |
|---|---|---|---|
| W1 | Steady | Constant rate | 5,000 events/s |
| W2 | Bursty | Baseline + 30s bursts | 2,000-15,000 events/s |
| W3 | Growing | Linear ramp over 5 min | 1,000-10,000 events/s |
| W4 | Shift | Throughput to latency-sensitive | 5,000 then 3,000 events/s |
| W5 | Misconfigured | W1 from degraded start | 5,000 events/s |
| ID | Name | Description |
|---|---|---|
| B0 | Misconfigured | Deliberately degraded settings (W5 only) |
| B1 | Default | Factory defaults for all three systems |
| B2 | Expert | Best-practice values from official documentation |
| B3 | Indep. BO | Per-component Bayesian optimization (Optuna TPE, 30 trials/system) |
| B4 | Joint BO | Joint 20-parameter Bayesian optimization (Optuna TPE, 100 trials) |
| B5 | PipeTune | LLM-driven cross-system tuning (proposed) |
# Full suite: 6 baselines x 4 workloads x 5 seeds = 120 runs
python run_experiments.py
# Single baseline or workload
python run_experiments.py --baseline B5
python run_experiments.py --workload W1
# Statistical analysis
python postprocess.py# Start multi-network testbed
docker compose -f docker-compose.distributed.yml up -d
# Inject 1ms inter-tier RTT via tc netem
./inject_latency.sh add
# Submit Flink job and run validation
bash run_dist_full.sh
# Teardown
docker compose -f docker-compose.distributed.yml down -vpython plots/generate_all.py
# Output: plots/figures/fig{1-6}_*.pdf| Component | Version |
|---|---|
| Apache Kafka | 3.6.1 |
| Apache Flink | 1.18.1 |
| PostgreSQL | 16.2 |
| Ollama | 0.5.4 |
| Llama 3.2 3B | Q4_K_M (2.0 GB) |
| Python | 3.10+ |
| Java | 11 (Flink job) |
Fixed random seeds: {42, 123, 456, 789, 1024}
| Container | vCPU | RAM |
|---|---|---|
| Kafka broker | 2 | 4 GB |
| Flink JobManager | 1 | 2 GB |
| Flink TaskManager | 2 | 4 GB |
| PostgreSQL | 2 | 4 GB |
| Ollama (LLM) | 2 | 6 GB |
| PipeTune agent | 1 | 2 GB |
| Total | 10 | 22 GB |
@inproceedings{inan2026pipetune,
author = {Umur Inan and Ata Turhan},
title = {{PipeTune}: {LLM}-Driven Cross-System Configuration Optimization
for Multi-Component Data Pipelines},
booktitle = {Proc. IEEE International Conference on Big Data (BigData)},
year = {2026}
}MIT