Verwenden des Autoloaders mit Unity Catalog

Der Autoloader kann Daten aus externen Speicherorten, die mit Unity Catalog konfiguriert sind, sicher erfassen. Weitere Informationen zum sicheren Verbinden von Speicher mit Unity-Katalog finden Sie unter Herstellen einer Verbindung mit dem Cloudobjektspeicher mithilfe des Unity-Katalogs. Der Autoloader basiert auf strukturiertem Streaming für die inkrementelle Verarbeitung. Empfehlungen und Informationen zu Einschränkungen finden Sie unter Verwenden von Unity Catalog mit strukturiertem Streaming.

Hinweis

In Databricks Runtime 11.3 LTS und höher können Sie auto Loader entweder mit Standard- oder dedizierten Zugriffsmodi (früher gemeinsam genutzte und Einzelbenutzerzugriffsmodi) verwenden.

Der Verzeichnisauflistungsmodus wird standardmäßig unterstützt.

Angeben von Speicherorten für Auto Loader-Ressourcen für Unity-Katalog

Das Unity Catalog-Sicherheitsmodell geht davon aus, dass alle Speicherorte, auf die in einer Workload verwiesen wird, von Unity Catalog verwaltet werden. Databricks empfiehlt, Prüfpunkt- und Schemaentwicklungsinformationen immer an Speicherorten zu speichern, die von Unity Catalog verwaltet werden. Unity Catalog ermöglicht es Ihnen nicht, Checkpoint-Dateien oder Dateien für Schema-Erkennung und -Entwicklung im Tabellenverzeichnis zu schachteln.

Aufnehmen von Daten aus cloudbasiertem Speicher mithilfe des Unity-Katalogs

In den folgenden Beispielen wird davon ausgegangen, dass der ausführende Benutzer READ FILES über Berechtigungen für den externen Speicherort, Besitzerberechtigungen für die Zieltabellen und die folgenden Konfigurationen und Berechtigungen verfügt.

Hinweis

Azure Data Lake Storage ist der einzige Azure-Speichertyp, der von Unity Catalog unterstützt wird.

Speicherort Erteilen
abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data READ FILES
abfss://dev-bucket@<storage-account>.dfs.core.windows.net READ FILES, WRITE FILESCREATE TABLE

Verwenden des automatischen Ladevorgangs zum Laden in eine verwaltete Unity-Katalog-Tabelle

Die folgenden Beispiele veranschaulichen, wie Sie das automatische Laden zum Aufnehmen von Daten in eine verwaltete Unity-Katalogtabelle verwenden.

Python

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))

SQL

CREATE OR REFRESH STREAMING TABLE dev_catalog.dev_database.dev_table
AS SELECT * FROM STREAM read_files(
  'abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data',
  format => 'json'
);

Wenn Sie read_files in einer CREATE STREAMING TABLE Anweisung innerhalb von Lakeflow-Pipelines verwenden, werden die Speicherorte für Prüfpunkte und Schemas automatisch verwaltet.

Verwenden Sie Auto Loader, um in eine externe Unity Catalog-Tabelle zu laden

Verwenden Sie eine externe Unity-Tabelle anstelle einer verwalteten Tabelle, um Daten an einem bestimmten Speicherort beizubehalten. Verwenden Sie beispielsweise eine externe Tabelle, um Daten für Nicht-Databricks-Clients freizugeben oder vorhandene Daten zu registrieren. Bei externen Tabellen legen Sie den Speicherpfad fest. Weitere Informationen finden Sie unter Arbeiten mit externen Tabellen.

Um das automatische Laden mit einer externen Tabelle des Unity Catalog zu verwenden, registrieren Sie zuerst die Tabelle mit CREATE TABLE ... LOCATION, und streamen Sie dann nach Namen in die Tabelle. Der Tabellenspeicherort muss sich an einem externen Speicherort befinden, an dem Sie über Berechtigungen verfügen CREATE EXTERNAL TABLE . Der Checkpoint-Speicherort muss auch in einem von Unity Catalog verwalteten externen Speicherort liegen. Verwenden Sie einen separaten Pfad von den Tabellendaten.

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"
table_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/external/dev_table"

# One-time: register the external table in UC.
spark.sql(f"""
  CREATE TABLE IF NOT EXISTS dev_catalog.dev_database.dev_table
  USING DELTA
  LOCATION '{table_path}'
""")

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))
  • Last updated on