Add Step DSL (Source/Flow/Merge/Split/Sink) for Spark jobs #313

Background

Workflow DSL (#310) covers job-to-job relationships only. The data flow inside a single Spark job needs its own abstraction. This task implements the Step model for Spark — Source / Flow / Merge / Split / Sink composition, both as a Scala DSL and as inline YAML via StepsRunnerJob. Other job kinds (e.g., bash) are out of scope.

Task

• Step sealed trait: Source (0→1) / Flow (1→1) / Merge (N→1) / Split (1→M) / Sink (1→0)
• Type-state Plan DSL: ~>, +, .as, fanOut, port selector for Split
• Executor: memoization, fanOut cache, Split port memo, plan-time validation
• StepsRunnerJob for inline YAML chains
• Built-ins: FileSource, SampleSource, SqlMerge, CacheFlow, FileSink, ShowSink
• Tests

Done When

• A Spark job can be expressed in Scala via Source ~> Flow ~> Sink with type-state safety (a chain without a Sink fails to compile).
• A Spark job can be expressed in YAML via kind: spark, mainClass: StepsRunnerJob, and an inline steps: list.
• Memoization is verifiable: a shared upstream feeding multiple sinks runs only once within a run.
• fanOut runs each branch against a cached upstream and unpersists after.
• A Split step produces M named outputs from one input, addressable via as: {port: label} (YAML) or forked("port") (Scala); the split body runs once per execution.
• All listed plan-time validations reject invalid graphs before execution.

Notes

Step signatures

A Merge receives labeled inputs, so consumers like SqlMerge use each label as the temp view name. The default label for a single-input chain is "_0". A Split declares output ports (string keys) addressed by downstream via as: {port: label} or forked("port").

Scala DSL

class JoinJob extends Job[JoinCfg] {
  override def plan(cfg: JoinCfg): Plan.Closed = {
    val users  = FileSource("users.parquet").as("u")
    val events = FileSource("events.parquet").as("e")

    (users + events) ~>
      SqlMerge("SELECT u.name, e.kind FROM u JOIN e ON u.id = e.user_id") ~>
      ShowSink()
  }
}

YAML inline — StepsRunnerJob

jobs:
  pi:
    kind: spark
    artifact: "com.kakao.actionbase:pipeline:0.x"
    mainClass: StepsRunnerJob
    args:
      steps:
        - step: SampleSource
          args: { n: 1000000, columns: [x, y] }
        - step: SqlMerge
          args:
            query: "SELECT 4.0*SUM(CASE WHEN x*x+y*y<=1 THEN 1 ELSE 0 END)/COUNT(*) AS pi FROM _0"
        - step: ShowSink

Related: #310 (workflow DSL), #311 (initial PR — closed in favor of split).