Workflow Spark Recommendation architecture

Architecture status

Label	Meaning
Implemented	Source exists at the recorded baseline
KCL-defined	Current platform-stacks renders the component or setting
Live-verified	Behavior was captured in gj76anadpsc, workspace w-t5e2kcxjd
Target	Required advisory architecture; not current behavior

The generation job, API, BFF routes, and Workflow presentation are implemented. The API and scheduled job are KCL-defined and enabled by default. Their current pods, schedule, table contents, request behavior, and role enforcement in the assessed realm still require live capture.

Current topology

Current KCL schedules one ScheduledSparkApplication daily at 0 18 * * * with concurrencyPolicy: Forbid. It provisions one on-demand driver and one spot executor. The API runs with two replicas and queries Delta through Trino.

Source map

Component	Evidence paths
Recommendation job/API	job/main.py, job/spark_fetch.py, job/llm_processor.py, job/delta_writer.py, job/retention.py, api/api.py in sparqd/spark-recommendation
Browser	RecommendationModal.tsx, Workflow list/detail hooks, and Spark-task serialization in sparqd/oqullus
BFF	sparkRecommendationRoute.js and sparkRecommendationService.js in sparqd/bff-api
Workflow contract	SparkKubernetesTask schema and Spark Kubernetes templates in sparqd/workflow-backend
Deployment	kcl/applications/spark_recommendation/main.k and Delta Spark flavor options in sparqd/platform-stacks

Current generation flow

1. Read the complete Spark History Server applications list.
2. Request environment metadata for each application sequentially.
3. Keep the latest completed application for each dag_id + task_id.
4. Query the complete resource-monitoring table and retain its latest row for each dag_id + task_id.
5. Join the two inputs by DAG/task name, without run or application identity.
6. Fetch Spark configuration, executor, job, and stage summaries.
7. Send one prompt per selected task to the AI gateway sequentially.
8. Parse three text sections: description, old configuration, and new configuration.
9. Merge by app_id + current run_date into a date-partitioned Delta table.
10. Rank expired records, but only log that retention was skipped.

The job has no processed-application checkpoint. If no newer Spark application exists, the same application can be analyzed and billed again on the next schedule.

Recommendation validity boundary

The current inputs do not support a reliable effectiveness claim:

• Spark History and monitoring records are independently selected as “latest” by DAG/task, so they can describe different executions.
• History Server totalDuration is stored as cpu_time_sec; it is task duration, not executor CPU utilization.
• maxMemory is capacity, but the implementation labels its maximum as peak_memory_used_mb.
• spark.executor.cores and executor memory are reported as cluster totals without multiplying by the executor count over time.
• resource metrics can be absent, yet the LLM still generates advice from the remaining input.
• the prompt constrains prose, but code performs no semantic validation of returned configuration keys or values.

The target input envelope is immutable and contains Workflow UUID and revision, workflow run ID, task ID and attempt, Spark application ID and attempt, application configuration fingerprint, Spark/runtime version, node-policy version, observation window, source completeness, and source timestamps.

Advisory applicability contract

Every suggested change must be structured rather than recovered from free text:

target field or Spark key
current value
suggested value
application phase: submission | pre-query runtime | informational
evidence and observation IDs
confidence and limitations
supported Cogrion field/path
policy-validation result
expected outcome and comparison metric

Submission-time settings map to the saved Workflow task and apply only to a future SparkApplication. Runtime SQL settings are shown only when the deployed Spark version marks them mutable and Cogrion exposes a supported configuration path. Generic node-family advice remains informational until it maps to an environment-approved driver_node_type or executor_node_type.

The current modal violates this contract by converting every returned line to spark.conf.set(key, value) Python. Workflow-backend already models resource fields and an arbitrary spark_conf map, but Oqullus does not currently send spark_conf for Spark tasks.

Storage and retention

Current Delta design

The table stores configuration maps, JSON summaries, LLM text, token counts, and estimated cost. Initial creation partitions by run_date; subsequent writes MERGE on app_id + run_date. Interactive reads normalize a DAG name, rank all matching history by run_ts, and return the latest row per task.

This is technically valid Delta usage, but poorly matched to small interactive state. Date partitioning creates many low-volume partitions while the primary read does not constrain date. Repeated updates create obsolete files. Delta documentation confirms that row deletion does not reclaim physical files; VACUUM is a separate, non-automatic operation and removes time-travel access beyond its retained boundary. Delta also provides compaction controls, but running compaction without enough files would add compute cost rather than solve a measured problem.

Current retention never reaches even logical deletion: the delete merge is commented out. RETENTION_RUNS only controls which old rows are counted and reported as skipped.

Recommended target

Use an operational store for latest advice, lifecycle state, user disposition, and bounded per-Workflow reads. Key it by exact task execution plus input and configuration fingerprints. Optionally append immutable generation and outcome events to Delta for fleet analysis, model evaluation, audit, and longer-term cost reporting.

If Delta remains the only store:

• do not partition a small table by daily date;
• append immutable generation records and materialize latest state separately;
• skip applications whose fingerprint was already processed;
• implement explicit logical retention and scheduled VACUUM with a documented time-travel boundary;
• compact only when file counts justify its compute cost; and
• predicate, paginate, limit, time out, and cancel Trino reads.

Security and access boundary

BFF validates the browser JWT but accepts caller-provided DAG names without resolving a managed Workflow permission. The recommendation API has no caller authentication and interpolates dagName into SQL text. Normalization is not sanitization; a quote in the name reaches the query string.

The target path is:

Oqullus
  → BFF validates user
  → workflow-backend resolves Workflow UUID and view permission
  → authenticated internal recommendation request by immutable identity
  → parameterized bounded query

Prompts can include Spark properties and infrastructure observations. The AI gateway contract must define allowed fields, redaction, model/data residency, logging, retention, and tenant isolation before production use.

Cost model

Cost grows with more than table size:

Boundary	Current scaling behavior
History discovery	Full application list plus an environment call per application
Metrics enrichment	Full latest-per-task window over the monitoring table, then all rows returned to the driver
LLM	One sequential call per selected task, including unchanged applications
Delta	Small daily writes, repeated merges, obsolete files, no cleanup
Read API	Two replicas and one Trino query per DAG request
Workflow list	BFF fans out one API request per visible Workflow

The stored text is small; compute and request amplification dominate. Record job duration, applications discovered/new/skipped, History requests, Trino scanned bytes, Delta files/bytes, LLM tokens/cost, API latency, and failed or invalid recommendation counts.

The stored llm_cost_usd is only an estimate: source applies one hard-coded $0.002 per 1,000 total tokens without identifying model price, separating input/output rates, or versioning the pricing basis. It must not be used as an authoritative chargeback value.

Failure and recovery

• A failed History endpoint is often converted to an empty summary, which can still be sent to the LLM.
• AI calls retry up to five times, including errors that may not be retryable.
• One task failure does not provide a persisted generation status for the UI.
• API health proves only process liveness, not Trino/table readiness.
• BFF bulk reads use Promise.allSettled, silently omitting failed DAGs.
• Oqullus has no unavailable, stale, invalid, or unsupported-recommendation state distinct from no recommendation.

Recovery must preserve generation state per exact application, retry only retryable failures with idempotency, expose last-good age without presenting it as current, and prevent incomplete inputs from producing actionable advice.

Delivery boundary

Stabilization should precede any apply capability:

1. correct metric semantics and execution identity;
2. validate a structured advisory schema against Workflow policy;
3. implement retention, bounded reads, authentication, and authorization;
4. measure recommendation acceptance and comparable-run outcomes; and
5. only then consider a reviewed “create Workflow draft” action.

No current assessment requirement authorizes automatic application.