Verwenden von Senken in Pipelines

Verwenden Sie die Lakeflow-Pipeline-API sink mit Flüssen , um Datensätze zu schreiben, die von einer Pipeline in eine externe Datensenke transformiert wurden. Zu den externen Datensenken gehören verwaltete und externe Tabellen des Unity-Katalogs sowie Ereignisstreamingdienste wie Apache Kafka oder Azure Event Hubs. Sie können auch Datensenken verwenden, um in benutzerdefinierte Datenquellen zu schreiben, indem Sie Python-Code für diese Datenquelle schreiben.

Eine Übersicht über Senkenkonzepte und deren Verwendung finden Sie unter Senken in Lakeflow-Pipelines.

Hinweis

Senken-Workflow

Wenn Ereignisdaten aus einer Streamingquelle in Ihre Pipeline aufgenommen werden, verarbeiten und verfeinern Sie diese Daten in Transformationen in Ihrer Pipeline. Anschließend verwenden Sie die Anfügeflow-Verarbeitung, um die transformierten Datensätze zu eine Senke zu streamen. Sie erstellen diese Spüle mit der create_sink()-Funktion. Weitere Informationen zur create_sink Funktion finden Sie in der Sink-API-Referenz.

Wenn Sie über eine Pipeline verfügen, die Ihre Datenstrom-Ereignisdaten erstellt oder verarbeitet und Datensätze für das Schreiben vorbereitet, können Sie eine Senke verwenden.

Die Implementierung einer Spüle besteht aus zwei Schritten:

  1. Erstellen Sie die Spüle.
  2. Verwenden Sie einen Anfügeflow oder einen Aktualisierungsflow, um die vorbereiteten Datensätze in die Senke zu schreiben.

Erstellen einer Spüle

Databricks unterstützt mehrere Arten von Zielsenken, in die Sie Ihre Datensätze schreiben, die aus Ihren Datenstromdaten verarbeitet wurden:

  • Delta-Tabellensenken (einschließlich verwalteter und externer Unity Catalog-Tabellen)
  • Apache Kafka-Senken
  • Azure Event Hubs-Senken
  • Benutzerdefinierte Senken, die in Python geschrieben wurden und auf benutzerdefinierten Python-Datenquellen basieren

Im Folgenden finden Sie Beispiele für Konfigurationen für Delta- und Kafka- und Azure Event Hub-Senken sowie benutzerdefinierte Python-Datenquellen:

Delta-Senken

So erstellen Sie eine Delta-Senke anhand des Dateipfads:

dp.create_sink(
  name = "delta_sink",
  format = "delta",
  options = {"path": "/Volumes/catalog_name/schema_name/volume_name/path/to/data"}
)

So erstellen Sie eine Delta-Senke anhand des Tabellennamens mithilfe eines vollqualifizierten Katalogs und Schemapfads:

dp.create_sink(
  name = "delta_sink",
  format = "delta",
  options = { "tableName": "catalog_name.schema_name.table_name" }
)

Kafka- und Azure Event Hubs-Senken

Dieser Code funktioniert sowohl für Apache Kafka- als auch für Azure Event Hubs-Senken.

credential_name = "<service-credential>"
eh_namespace_name = "dp-eventhub"
bootstrap_servers = f"{eh_namespace_name}.servicebus.windows.net:9093"
topic_name = "dp-sink"

dp.create_sink(
name = "eh_sink",
format = "kafka",
options = {
    "databricks.serviceCredential": credential_name,
    "kafka.bootstrap.servers": bootstrap_servers,
    "topic": topic_name
  }
)

Dies credential_name ist ein Verweis auf ein Dienstberechtigungsnachweis des Unity-Katalogs. Weitere Informationen finden Sie unter Verwenden von Unity Catalog-Dienstanmeldeinformationen zum Herstellen einer Verbindung mit externen Clouddiensten.

Benutzerdefinierte Python-Datenquellen

Wenn Sie eine benutzerdefinierte Python-Datenquelle als my_custom_datasourceregistriert haben, kann der folgende Code in diese Datenquelle schreiben.

from pyspark import pipelines as dp

# Assume `my_custom_datasource` is a custom Python streaming
# data source that writes data to your system.

# Create Lakeflow pipelines sink using my_custom_datasource
dp.create_sink(
    name="custom_sink",
    format="my_custom_datasource",
    options={
        <options-needed-for-custom-datasource>
    }
)

# Create append flow to send data to RequestBin
@dp.append_flow(name="flow_to_custom_sink", target="custom_sink")
def flow_to_custom_sink():
    return read_stream("my_source_data")

Ausführliche Informationen zum Erstellen benutzerdefinierter Datenquellen in Python finden Sie unter PySpark benutzerdefinierte Datenquellen.

Weitere Informationen zur Verwendung der create_sink-Funktion finden Sie in der Senken-API-Referenz.

Nachdem Ihre Senke erstellt wurde, können Sie mit dem Streamen von verarbeiteten Datensätzen in die Senke beginnen.

Schreiben in eine Senke mit einem Anfügeflow

Nachdem die Senke erstellt wurde, besteht der nächste Schritt darin, verarbeitete Datensätze in sie zu schreiben, indem sie als Ziel für Datensätze angegeben wird, die von einem Anfügefluss ausgegeben werden. Dazu geben Sie Ihre Senke als target-Wert im append_flow-Decorator an.

  • Verwenden Sie für verwaltete und externe Tabellen im Unity-Katalog das Format delta, und geben Sie den Pfad- oder Tabellennamen in Den Optionen an. Ihre Pipeline muss für die Verwendung des Unity-Katalogs konfiguriert sein.
  • Verwenden Sie für Apache Kafka-Themen das Format kafka, und geben Sie den Themennamen, Verbindungsinformationen und Authentifizierungsinformationen in den Optionen an. Dies sind die gleichen Optionen, die ein Spark Structured Streaming Kafka sink unterstützt. Siehe Konfigurieren des Kafka Structured Streaming Writer.
  • Verwenden Sie für Azure Event Hubs das Format kafka, und geben Sie den Namen, die Verbindungs- und Authentifizierungsinformationen in den Optionen an. Dies sind die gleichen Optionen, die in einer Spark Structured Streaming Event Hubs-Spüle unterstützt werden, die die Kafka-Schnittstelle verwendet. Siehe Authentifizierung.

Im Folgenden finden Sie Beispiele für das Einrichten von Flows zum Schreiben in Delta-, Kafka- und Azure Event Hubs-Senken mit Datensätzen, die von Ihrer Pipeline verarbeitet werden.

Delta-Spüle

@dp.append_flow(name = "delta_sink_flow", target="delta_sink")
def delta_sink_flow():
  return(
  spark.readStream.table("spark_referrers")
  .selectExpr("current_page_id", "referrer", "current_page_title", "click_count")
)

Kafka- und Azure Event Hubs-Senken

@dp.append_flow(name = "kafka_sink_flow", target = "eh_sink")
def kafka_sink_flow():
return (
  spark.readStream.table("spark_referrers")
  .selectExpr("cast(current_page_id as string) as key", "to_json(struct(referrer, current_page_title, click_count)) AS value")
)

Der value-Parameter ist für eine Azure Event Hubs-Senke obligatorisch. Zusätzliche Parameter wie key, partition, headersund topic sind optional.

Weitere Details zum append_flow-Decorator finden Sie unter Standard-Flows und angehängte Flows.

Einschränkungen

  • Nur die Python-API wird unterstützt. SQL wird nicht unterstützt.

  • Nur Streamingabfragen werden unterstützt. Batchabfragen werden nicht unterstützt.

  • Nur append_flow und update_flow können verwendet werden, um in Senken zu schreiben. Andere Flow-Prozesse, wie create_auto_cdc_flow, werden nicht unterstützt, und Sie können keine Senke in der Definition eines Pipeline-Datasets verwenden. Beispielsweise wird Folgendes nicht unterstützt:

    @table("from_sink_table")
def fromSink():
  return read_stream("my_sink")

  • Für Delta-Senken muss der Tabellenname vollqualifiziert sein. Insbesondere muss der Tabellenname für im Unity-Katalog verwaltete externe Tabellen das Format <catalog>.<schema>.<table>haben. Für den Hive-Metastore muss sie in der Form „<schema>.<table>“ sein.

  • Das Ausführen vollständigen Aktualisierungsupdates bereinigt zuvor berechnete Ergebnisdaten in den Senken nicht. Dies bedeutet, dass alle verarbeiteten Daten an die Senke angefügt werden und vorhandene Daten nicht geändert werden.

  • Pipelineerwartungen werden nicht unterstützt.

  • Die serverlose Egress-Steuerung unterstützt nur Kafka- und Delta Lake-Senken-Connectors. Siehe Was ist serverlose Ausgangskontrolle?

Weitere Ressourcen

  • Last updated on