Verwenden von Parametern mit Pipelines

Mithilfe von Pipelineparametern können Sie den gleichen Pipelinequellcode in Umgebungen oder Datasets wiederverwenden. Sie können z. B. dieselben Transformationen auf die Kataloge dev und prod anwenden oder bei jedem Durchlauf Daten aus einem anderen Quellpfad einlesen. Sie definieren Parameter in der Pipeline (oder überschreiben sie beim Starten einer Aktualisierung) und referenzieren sie in Ihrem SQL-Quellcode.

Auf dieser Seite wird das Pipelineparameterfeature beschrieben, das für SQL-Quellcode verfügbar ist. Um Python Quellcode in einer Pipeline zu parametrisieren, verwenden Sie weiterhin das Feld Configuration wie in Reference-Parametern mithilfe des Konfigurationsfelds beschrieben. Die Konfiguration wird auch verwendet, um Spark-Konfigurationswerte festzulegen, die Pipelines zur Laufzeit lesen. Ausführliche Informationen zu den Spark-Konfigurationseinstellungen finden Sie in der Referenz zu Pipelineeigenschaften.

Was sind Pipelineparameter?

Pipelineparameter sind Schlüsselwertpaare, die Sie verwenden können:

• Legen Sie dies in den Pipeline-Einstellungen als Standard fest.
• Beim Starten eines Updates über die Pipeline-Benutzeroberfläche, die Start update API oder das Dialogfeld „Mit anderen Einstellungen ausführen“ überschreiben.
• Überschreiben Sie die Pipelineaufgabe in einem Auftrag mit optionalem Pushdown von Parametern auf Auftragsebene.
• Verweis aus dem SQL-Quellcode unter Verwendung der Syntax für benannte Parameter.

Parameterwerte sind immer Zeichenfolgen. Schlüssel können alphanumerische Zeichen, Unterstriche (_), Bindestriche (-) und Punkte (.) enthalten.

Pipelineparameter und das Feld "Konfiguration " dienen verschiedenen Zwecken:

Definieren von Pipelineparametern

Sie können Standardparameterwerte in Pipelineeinstellungen definieren. Wenn ein Update ohne Außerkraftsetzungen ausgeführt wird, verwendet die Pipeline diese Standardwerte.

Verwenden Sie die Pipeline-Benutzeroberfläche

1. Klicken Sie in Ihrem Arbeitsbereich auf Aufträge und Pipelines in der Randleiste, und wählen Sie Ihre Pipeline aus.
2. Klicken Sie auf Einstellungen.
3. Suchen Sie in der Randleiste der Pipelineeinstellungen den Abschnitt "Parameter ", und klicken Sie auf "Bearbeiten".
4. Fügen Sie Schlüssel - und Werteinträge hinzu, und klicken Sie dann auf "Speichern".

Verwenden der JSON- oder REST-API

Fügen Sie der Pipelinedefinition ein parameters Mapping hinzu:

{
  "name": "Sales pipeline",
  "parameters": {
    "source_catalog": "dev_catalog",
    "source_schema": "sales",
    "start_date": "2026-01-01"
  }
}

Die vollständige JSON-Referenz zur Pipeline finden Sie unter Pipelinekonfigurationen.

Referenzparameter im SQL-Quellcode

Verweisen Sie auf einen Parameter, indem Sie dem Schlüssel einen Doppelpunkt voranstellen. Azure Databricks bindet den Wert zur Aktualisierungszeit als Zeichenfolge:

CREATE OR REFRESH MATERIALIZED VIEW transaction_summary AS
SELECT account_id,
  COUNT(txn_id) AS txn_count,
  SUM(txn_amount) AS account_revenue
FROM :source_catalog.sales.transactions
WHERE txn_date >= :start_date
GROUP BY account_id

Um einen Parameter an einer Bezeichnerposition zu verwenden, z. B. einen Katalog, ein Schema oder einen Tabellennamen, umschließen Sie ihn in IDENTIFIER():

USE CATALOG IDENTIFIER(:source_catalog);
USE SCHEMA IDENTIFIER(:source_schema);

CREATE OR REFRESH MATERIALIZED VIEW daily_sales AS
SELECT date(timestamp) AS date,
  SUM(price) AS total_sales
FROM transactions
GROUP BY date;

Wenn der Quellcode auf einen Parameter verweist, der zur Aktualisierungszeit keinen Wert aufweist, schlägt das Update mit einem Fehler fehl. Die Pipeline ignoriert zusätzliche Parameter, auf die der Code nicht verweist.

Parameter beim Aktualisieren überschreiben

Sie können Parameterwerte für ein einzelnes Update außer Kraft setzen, ohne die gespeicherten Standardwerte zu ändern.

• Klicken Sie auf der Pipeline-Benutzeroberfläche auf "Ausführen" mit unterschiedlichen Einstellungen , und bearbeiten Sie den Abschnitt "Parameter ".
• Legen Sie in einer Pipelineaufgabe in einem Auftrag Parameterüberschreibungen im Feld Parameters der Aufgabe fest. Siehe Parameter.
• Übergeben Sie aus der API eine parameters Karte in der Startupdateanforderung .

Azure Databricks zeichnet die Parameter für ein bestimmtes Update im Updateverlauf auf und zeigt sie in der Run-ParameterSpalte der Pipelineausführungsliste an.

Rangfolge des Parameters

Wenn Sie denselben Schlüssel an mehreren Stellen definieren, gewinnt der Wert mit der höchsten Rangfolge. Von hoch nach niedrig:

1. Parameter für die Auftragsausführung: Werte, die für eine einzelne Auftragsausführung angegeben werden (Überschreibungen).
2. Auftragsparameter: Standardwerte, die für den übergeordneten Auftrag definiert sind.
3. Parameter für Pipelineaufgaben: Werte, die für den Pipelinevorgang festgelegt sind.
4. Pipelineparameter: Standardwerte, die in Pipelineeinstellungen definiert sind.

Dies entspricht der Rangfolge, die von anderen Auftragsparameteraufgabentypen verwendet wird.

Parameter für Pipelines in Lakeflow Jobs

Wenn Sie eine Pipeline als Pipelineaufgabe in einen Job einplanen, kann die Aufgabe Parameter bereitstellen, die die Standardwerte der Pipeline überschreiben. Parameterwerte können dynamische Wertverweise verwenden, um Werte zur Auftragslaufzeit wie {{job.trigger.time.iso_date}} oder {{job.parameters.region}} einzufügen.

Lakeflow Jobs geben außerdem alle Job-Parameter automatisch an Pipeline-Aufgaben weiter, genauso wie an Notebook- und SQL-Aufgaben. Der Quellcode der Pipeline kann mithilfe der Syntax benannter Parameter auf jeden übergebenen Wert verweisen. Das Deklarieren eines Parameters in pipelineeinstellungen ist optional und legt nur einen Standardwert für die Ausführung ohne Außerkraftsetzung fest.

Wichtige Hinweise und bekannte Einschränkungen

• Pipelines führen jeweils ein Update aus: Eine Pipeline kann jeweils nur ein einzelnes Update ausführen. Um zu verhindern, dass Aufträge fehlschlagen, wenn sich mehrere Updates andernfalls überschneiden würden, begrenzt Azure Databricks die Parallelität in zwei Szenarien auf 1:

  • Ein Auftrag, der eine Pipelineaufgabe enthält und mit mehr als einem max_concurrent_runs konfiguriert ist.
  • Eine Pipelineaufgabe, die in eine For-Each-Task eingebettet ist, unabhängig von der Anzahl der Iterationen.

  Die Auftrags-UI zeigt eine Benachrichtigung an, wenn diese Begrenzung greift. Berücksichtigen Sie die Obergrenze beim Entwurf parametrisierter Pipelines, die Sie mit vielen Parameterkombinationen ausführen möchten.

• Datumsfilter können vollständige Aktualisierungen auslösen: Ein gängiger Parameteranwendungsfall besteht darin, Daten nach Datum zu filtern. Vorsicht bei Prädikaten: Wenn auf beiden Seiten eines Datumsbereichs gefiltert wird, wird die inkrementelle Verarbeitung bei materialisierten Ansichten verhindert und bei jeder Aktualisierung eine vollständige Aktualisierung ausgelöst.

  -- Triggers a full refresh on each update
  CREATE OR REFRESH MATERIALIZED VIEW recent_orders AS
  SELECT * FROM orders
  WHERE order_date >= :start_date AND order_date < :end_date;
  

  -- Processes incrementally
  CREATE OR REFRESH MATERIALIZED VIEW recent_orders AS
  SELECT * FROM orders
  WHERE order_date >= :start_date;
  

• Benannte Parameter sind nur in SQL verfügbar: In dieser Betaversion kann die Syntax für benannte Parameter nur im SQL-Quellcode verwendet werden. Um Python Quellcode zu parametrisieren, verwenden Sie weiterhin das Feld Configuration mit spark.conf.get(). Siehe Referenzparameter mithilfe des Konfigurationsfelds.

Parameter mithilfe des Konfigurationsfelds referenzieren

Das Feld "Konfiguration" in einer Pipeline akzeptiert beliebige Schlüsselwertpaare, die als Spark-Konfigurationswerte verfügbar gemacht werden. Dies ist der alte Parametrisierungsmechanismus und funktioniert weiterhin neben Pipeline-Parametern. Verwenden Sie ihn für Python Quellcode und für Schlüssel, die Sie mit spark.conf.get() lesen möchten, anstatt für benannte Parametersyntax.

Im folgenden Beispiel wird ein mypipeline.start_date Konfigurationswert verwendet, um eine Entwicklungspipeline auf eine Teilmenge von Eingabedaten zu beschränken:

SQL

CREATE OR REFRESH MATERIALIZED VIEW customer_events
AS SELECT * FROM source_table WHERE date > '${mypipeline.start_date}';

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table
def customer_events():
  start_date = spark.conf.get("mypipeline.start_date")
  return spark.read.table("source_table").where(col("date") > start_date)

Sie legen Konfigurationswerte im Konfigurationsabschnitt der Pipelineeinstellungen oder im configuration Feld der Pipeline-JSON fest. Vermeiden Sie Schlüssel, die mit reservierten Pipeline- oder Apache Spark-Konfigurationswerten in Konflikt geraten.

Verwandte Artikel

• Konfigurieren von Pipelines
• Pipelineaufgabe für Jobs
• Konfigurieren von Auftragsparametern
• Was ist ein dynamischer Wertverweis?