# `SparkEx.StreamWriter`
[🔗](https://github.com/lukaszsamson/spark_ex/blob/v0.1.1/lib/spark_ex/stream_writer.ex#L1)

Builder API for starting structured streaming queries.

Mirrors PySpark's `DataStreamWriter` with a builder pattern.

## Examples

    df
    |> SparkEx.DataFrame.write_stream()
    |> SparkEx.StreamWriter.format("console")
    |> SparkEx.StreamWriter.output_mode("append")
    |> SparkEx.StreamWriter.start()

# `t`

```elixir
@type t() :: %SparkEx.StreamWriter{
  cluster_by: [String.t()],
  df: SparkEx.DataFrame.t(),
  foreach_batch: SparkEx.Types.foreach_function() | nil,
  foreach_writer: SparkEx.Types.foreach_function() | nil,
  options: %{required(String.t()) => String.t()},
  output_mode: String.t() | nil,
  partition_by: [String.t()],
  path: String.t() | nil,
  query_name: String.t() | nil,
  source: String.t() | nil,
  table_name: String.t() | nil,
  trigger: term() | nil
}
```

# `cluster_by`

```elixir
@spec cluster_by(t(), [String.t()]) :: t()
```

# `foreach_batch`

```elixir
@spec foreach_batch(t(), SparkEx.Types.foreach_function()) :: t()
```

Sets a foreach batch function for micro-batch processing.

Accepts a Spark Connect StreamingForeachFunction proto struct.
Use this with Java/Scala UDF payloads.

## Example with Scala UDF

    foreach_fn = %Spark.Connect.StreamingForeachFunction{
      function: {:scala_function, %Spark.Connect.ScalarScalaUDF{
        payload: serialized_scala_bytes
      }}
    }
    writer |> StreamWriter.foreach_batch(foreach_fn)

# `foreach_writer`

```elixir
@spec foreach_writer(t(), SparkEx.Types.foreach_function()) :: t()
```

Sets a foreach writer function for row-level processing.

Accepts a Spark Connect StreamingForeachFunction proto struct.
Use this with Java/Scala UDF payloads.

## Example with Scala UDF

    foreach_fn = %Spark.Connect.StreamingForeachFunction{
      function: {:scala_function, %Spark.Connect.ScalarScalaUDF{
        payload: serialized_scala_bytes
      }}
    }
    writer |> StreamWriter.foreach_writer(foreach_fn)

# `format`

```elixir
@spec format(t(), String.t()) :: t()
```

# `option`

```elixir
@spec option(t(), String.t(), term()) :: t()
```

# `options`

```elixir
@spec options(t(), map() | keyword()) :: t()
```

# `output_mode`

```elixir
@spec output_mode(t(), String.t()) :: t()
```

# `partition_by`

```elixir
@spec partition_by(t(), [String.t()]) :: t()
```

# `path`

```elixir
@spec path(t(), String.t()) :: t()
```

Sets the output path for the streaming sink.

# `query_name`

```elixir
@spec query_name(t(), String.t()) :: t()
```

# `start`

```elixir
@spec start(
  t(),
  keyword()
) :: {:ok, SparkEx.StreamingQuery.t()} | {:error, term()}
```

Starts the streaming query, writing to the path set via `option("path", ...)`.

Returns `{:ok, StreamingQuery.t()}` on success.

# `to_table`

```elixir
@spec to_table(t(), String.t(), keyword()) ::
  {:ok, SparkEx.StreamingQuery.t()} | {:error, term()}
```

Starts the streaming query, writing to the given table name.

Returns `{:ok, StreamingQuery.t()}` on success.

# `trigger`

```elixir
@spec trigger(
  t(),
  keyword()
) :: t()
```

Sets the trigger for the streaming query.

## Trigger types

- `processing_time: "5 seconds"` — micro-batch at the given interval
- `available_now: true` — process all available data then stop
- `once: true` — process one micro-batch then stop
- `continuous: "1 second"` — continuous processing at the given checkpoint interval

# `xml`

```elixir
@spec xml(t(), String.t()) :: t()
```

Starts the streaming query writing XML to the given path.

---

*Consult [api-reference.md](api-reference.md) for complete listing*