create_sink

La funzione create_sink() scrive in un servizio di streaming di eventi, ad esempio Apache Kafka o Hub eventi di Azure o in una tabella Delta da una pipeline dichiarativa. Dopo aver creato un sink con la create_sink() funzione, usare il sink in un flusso di accodamento o un flusso di aggiornamento per scrivere dati nel sink. I flussi di accodamento e aggiornamento sono gli unici tipi di flusso supportati con la create_sink() funzione . Altri tipi di flusso, ad esempio create_auto_cdc_flow, non sono supportati. Per informazioni dettagliate su altri tipi di sink nelle pipeline di Lakeflow, vedere Sink nelle pipeline di Lakeflow.

Il sink Delta supporta le tabelle esterne e gestite di Unity Catalog e le tabelle gestite del metastore Hive. I nomi delle tabelle devono essere completamente qualificati. Ad esempio, le tabelle del catalogo Unity devono usare un identificatore a tre livelli: <catalog>.<schema>.<table>. Le tabelle del metastore Hive devono usare <schema>.<table>.

Annotazioni

  • L'esecuzione di un aggiornamento completo non cancella i dati dai raccoglitori. Tutti i dati elaborati vengono aggiunti al sink e i dati esistenti non vengono modificati.
  • Le aspettative non sono supportate con l'API sink .

Sintassi

from pyspark import pipelines as dp

dp.create_sink(name=<sink_name>, format=<format>, options=<options>)

Parametri

Parametro TIPO Description
name str Obbligatorio. Stringa che identifica il sink e viene utilizzata per fare riferimento e gestire lo sink. I nomi del sink devono essere univoci per la pipeline, inclusi tutti i file di codice sorgente che fanno parte della pipeline.
format str Obbligatorio. Stringa che definisce il formato di output, kafka o delta.
options dict Elenco di opzioni sink, formattate come {"key": "value"}, in cui la chiave e il valore sono entrambe stringhe. Sono supportate tutte le opzioni di Databricks Runtime supportate dai sink Kafka e Delta.

Esempi

from pyspark import pipelines as dp

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Create an external Delta table sink with a file path
dp.create_sink(
  "my_delta_sink",
    "delta",
    { "path": "/path/to/my/delta/table" }
)

# Create a Delta table sink using a table name
dp.create_sink(
  "my_delta_sink",
    "delta",
    { "tableName": "my_catalog.my_schema.my_table" }
)
  • Last updated on