Documentation

Website ↗ GitHub ↗

On this page

Sink

Sink¶

write_target.type: sink. sink_type selects the destination. Every sink emits dp.create_sink(name=, format=, options=) plus one @dp.append_flow(target=<sink_name>, name="f_<sink_name>_<index>", comment=) per source view, reading the source with spark.readStream.table(...).

Minimal example¶

Add this entry to the actions list in your flowgroup.

- name: write_orders_to_delta_sink
  type: write
  source: v_orders
  write_target:
    type: sink
    sink_type: delta
    sink_name: orders_delta_sink
    options:
      tableName: "${catalog}.${gold_schema}.orders_export"

Options¶

Key	Type	Default	Notes
`sink_type`	string	required	`delta`, `kafka`, `foreachbatch`, or `custom`.
`sink_name`	string	—	Unique sink identifier; used in the emitted flow names.
`options`	dict	`{}`	Sink options.
`comment`	string	derived	—

sink_type-specific fields¶

sink_type	Additional fields	Notes
`delta`	`options.tableName` or `options.path`	Format fixed `delta`; `tableName` and `path` mutually exclusive.
`kafka`	`bootstrap_servers`, `topic`, `options`	Format fixed `kafka`; source must carry a `value` column.
`kafka` (Event Hubs)	`options.kafka.sasl.mechanism: OAUTHBEARER`	No separate `sink_type`; `OAUTHBEARER` flips the Kafka handler into Event Hubs mode. Endpoint `<namespace>.servicebus.windows.net:9093`, Event Hub name as `topic`.
`foreachbatch`	`sink_name` (required), `module_path` or `batch_handler`	Exactly one of `module_path` / `batch_handler`; `source` must be a single view string.
`custom`	`module_path` (required), `custom_sink_class` (required), `options`	Writes through a user-supplied PySpark `DataSink`.

- name: write_orders_to_kafka_sink
  type: write
  source: v_orders_for_kafka
  write_target:
    type: sink
    sink_type: kafka
    sink_name: order_events_kafka
    bootstrap_servers: "${kafka_bootstrap_cluster}"
    topic: "acme.orders.fulfillment"
    options:
      kafka.security.protocol: "SASL_SSL"
      kafka.sasl.mechanism: "PLAIN"

- name: write_orders_to_eventhubs
  type: write
  source: v_orders_for_eventhubs
  write_target:
    type: sink
    sink_type: kafka
    sink_name: order_events_eventhubs
    bootstrap_servers: "${eh_namespace}.servicebus.windows.net:9093"
    topic: "acme-orders"
    options:
      kafka.security.protocol: "SASL_SSL"
      kafka.sasl.mechanism: "OAUTHBEARER"

- name: merge_customer_updates
  type: write
  source: v_customer_changes
  write_target:
    type: sink
    sink_type: foreachbatch
    sink_name: customer_merge_sink
    batch_handler: |
      df.createOrReplaceTempView("batch_view")
      df.sparkSession.sql("MERGE INTO ${catalog}.${gold_schema}.dim_customer ...")

- name: write_to_custom_sink
  type: write
  source: v_seed_rows
  write_target:
    type: sink
    sink_type: custom
    sink_name: backed_sink
    module_path: "py_functions/custom_sink.py"
    custom_sink_class: "MyCustomSink"
    options:
      output_path: "/tmp/custom_sink_output"

Action fields¶

These fields sit on the action, alongside write_target.

Field	Type	Default	Requirement and behaviour
`name`	string	required	Action name, unique within the flowgroup.
`type`	string	required	Always `write`.
`source`	string / list	conditional	Input view(s). Standard streaming writes and most sinks accept a list. CDC, replace, replace_where and foreachbatch use one view. Snapshot CDC takes its source inside `snapshot_cdc_config`. Materialized views require one of action `source`, target `sql` or target `sql_path`.
`write_target`	mapping	required	Target settings; `type` selects `streaming_table`, `materialized_view` or `sink`.
`description`	string	derived	Comment/docstring for the generated flow or view.
`readMode`	string	`stream`	`stream` or `batch` for standard streaming-table append flows. Replace requires streaming; replace_where reads in batch and rejects `stream`. Materialized views use batch queries; sinks use streaming reads. The read must match the kind of the view it reads, or `lhp validate` fails with `LHP-VAL-070`; a batch read in a standard streaming-table flow also needs `once: true`.
`once`	bool	`false`	Emits `once=True` for standard append and CDC flows. Not emitted for snapshot CDC, replace, replace_where, materialized views or sinks.
`operational_metadata`	bool / list	—	Accepted on the model; writes do not add metadata columns. Select columns on load or supported transform actions instead.
`depends_on`	list[string]	—	Extra upstream table/view references. See Dependency analysis reference.