Ładowanie danych w potokach

Dane można ładować z dowolnego źródła danych obsługiwanego przez platformę Apache Spark w usłudze Azure Databricks przy użyciu potoków. Zestawy danych (tabele i widoki) można zdefiniować w potoku względem dowolnego zapytania zwracającego ramkę danych platformy Spark, w tym przesyłania strumieniowego ramek danych i biblioteki Pandas dla ramek danych platformy Spark. Dla procesów integracyjnych danych usługa Databricks zaleca używanie tabel przesyłania strumieniowego w większości przypadków użycia. Tabele przepływu strumieniowego są przydatne do pozyskiwania danych z magazynu obiektów w chmurze przy użyciu Auto Loader lub z kolejek wiadomości, takich jak Kafka. Więcej informacji o tabelach strumieniowych, podstawowym typie zbioru danych używanym do pozyskiwania danych, można znaleźć w sekcji Tabele strumieniowe.

Nie wszystkie źródła danych mają wsparcie SQL do przetwarzania danych wejściowych. Można jednak mieszać źródła SQL i Python w tym samym potoku, aby używać Python tam, gdzie jest to konieczne. Aby uzyskać szczegółowe informacje na temat pracy z bibliotekami, które nie są domyślnie pakowane z potokami, zobacz Manage Python dependencies for pipelines (Zarządzanie zależnościami Python dla potoków). Aby uzyskać ogólne informacje na temat pozyskiwania danych w usłudze Azure Databricks, zobacz Łączniki standardowe w programie Lakeflow Connect.

W poniższych przykładach przedstawiono niektóre typowe wzorce ładowania danych.

Zidentyfikuj źródła danych i ścieżkę połączenia

Zanim napiszesz kod pipeline, zapisz każde miejsce, z którego pochodzą dane. Dla każdego źródła zwróć uwagę, jak udostępnia swoje dane (pliki, bazę danych, system oprogramowania jako usługi (SaaS), API lub strumień), jak często się zmienia oraz jakie dane uwierzytelniające i dostęp do sieci potrzebuje. Metoda połączenia często decyduje o tym, czy źródło jest z natury wsadowe czy strumieniowe, więc właściwe ustalenie tego na wczesnym etapie pozwala uniknąć późniejszej ponownej pracy.

Każde źródło przypisuj do jednej z następujących ścieżek połączeń. Poniższa tabela przedstawia preferowany mechanizm dla każdego źródła potoku:

Source Ścieżka połączenia
Pliki trafiają do chmurowej pamięci obiektowej (S3, Azure Data Lake Storage, GCS) Najczęstszy punkt wyjścia. Użyj AutoLoader (cloudFiles format), który obsługuje odkrywanie inkrementalne, wnioskowanie schematów oraz ewolucję schematu. Zobacz Ładowanie plików z magazynu obiektów w chmurze.
Bazy danych i aplikacje SaaS (Salesforce, SQL Server, PostgreSQL, Workday) Użyj zarządzanego złącza Lakeflow Connect tam, gdzie taki jest dostępny dla źródła. Zarządzane złącza są oparte na konfiguracji i automatycznie obsługują uwierzytelnianie oraz przyrostowe pozyskiwanie danych lub CDC. Zobacz Łączniki zarządzane w programie Lakeflow Connect. Jeśli dla Twojego źródła nie istnieje zarządzany konektor, pozyskaj dane bezpośrednio lub najpierw zapisz odpowiedzi jako pliki. Zobacz Pobieranie danych z API w potokach.
Szyny komunikatów (Kafka, Kinesis, Azure Event Hubs, Pub/Sub) Czytam bezpośrednio jako źródło streamingu strukturalnego, ponieważ są to natywne źródła streamingu. Zobacz Ładowanie danych z magistrali komunikatów.
Inne tabele Delta lub zasoby Unity Catalog, w tym tabele tworzone przez inne potoki danych lub zadania Odwołuj się do nich bezpośrednio i pozwól, aby mechanizmy zarządzania oraz śledzenia pochodzenia danych w Unity Catalog obsługiwały wyszukiwanie i dostęp. Zobacz Ładowanie z istniejącej tabeli.
Małe lub statyczne dane referencyjne (pliki wyszukiwania, rzadko zmieniające się CSV) Załaduj jako źródło wsadowe w widoku materializowanym. Nie ma żadnej korzyści ze streamowania czegoś, co prawie się nie zmienia. Zobacz Załaduj małe lub statyczne zbiory danych z chmurowego przechowywania obiektów.
Dowolne API HTTP lub REST bez zarządzanego złącza Pobieraj z API w pipeline lub najpierw wyląduj jego odpowiedzi jako pliki. Zobacz Pobieranie danych z API w potokach.

Dla każdego źródła potwierdź następujące kwestie przed rozpoczęciem kompilacji:

  • Tożsamość: Jak działa pipeline. Potoki mogą działać przy użyciu nazwy głównej usługi, więc najpierw to skonfiguruj, aby uniknąć zależności od konta osobistego.
  • Ścieżka sieci: Jakiej łączności potrzebuje źródło, na przykład dane uwierzytelniające do przechowywania danych, zewnętrzna lokalizacja lub zarządzana łączność Lakeflow Connect.
  • Semantyka zmian: To, jak źródło sygnalizuje aktualizacje i usunięcia, jeśli w ogóle. To decyduje, czy potrzebujesz CDC, czy możesz traktować źródło tylko jako dołączone.

Wybierz format pliku i warstwę pamięci

Rurociągi podejmują większość decyzji za ciebie. Każda tabela strumieniowa i zmaterializowany widok tworzony przez pipeline jest domyślnie przechowywany jako tabela Delta, co daje transakcje ACID, egzekwowanie i ewolucję schematu, podróże w czasie oraz zarządzanie i linię Unity Catalog na każdym zbiorze danych. Nie wybierasz formatu dla wyjść potoków. Twoje prawdziwe decyzje są na dwóch krańcach procesu:

  • Surowy format wejściowy: To, co generuje źródło, takie jak CSV, JSON lub Parquet. Auto Loader i read_files() obsługują je bezpośrednio. Określ format za pomocą elementu cloudFiles.format w Pythonie lub argumentu format => w SQL. Jeśli kontrolujesz źródło, wolej Parquet lub Avro, bo lepiej przenoszą schemat i kompresują, co przyspiesza pobieranie i wnioskowanie schematów. Potoki obsługują każdy z tych formatów, więc nie pozwól, by format ograniczał wybór źródła.
  • Lokalizacja surowych danych: W przypadku plików zapisuj dane w woluminie Unity Catalog, a nie w niezarządzanej ścieżce zasobnika, tak aby śledzenie pochodzenia danych i kontrola dostępu obejmowały także strefę przyjęcia danych. Zobacz Czym są wolumeny Unity Catalog?.

Dla tabel generowanych przez pipeline pozostałe opcje to katalog docelowy i schemat, które wyznaczają granice zarządzania i możliwość wykrywania, oraz fizyczny układ dużych tabel. Użyj CLUSTER BY (liquid clustering), aby utrzymać wysoką wydajność zapytań w miarę rozrastania się tabel bez ręcznego dostrajania partycji. Zobacz Używaj płynnego grupowania dla tabel.

Ładowanie z istniejącej tabeli

Załaduj dane z dowolnej istniejącej tabeli w usłudze Azure Databricks. Dane można przekształcić przy użyciu zapytania lub załadować tabelę dla dalszego przetwarzania w ramach kanału przetwarzania.

Python

@dp.table(
  comment="A table summarizing counts of the top baby names for New York for 2021."
)
def top_baby_names_2021():
  return (
    spark.read.table("baby_names_prepared")
      .filter(expr("Year_Of_Birth == 2021"))
      .groupBy("First_Name")
      .agg(sum("Count").alias("Total_Count"))
      .sort(desc("Total_Count"))
  )

SQL

CREATE OR REFRESH MATERIALIZED VIEW top_baby_names_2021
COMMENT "A table summarizing counts of the top baby names for New York for 2021."
AS SELECT
  First_Name,
  SUM(Count) AS Total_Count
FROM baby_names_prepared
WHERE Year_Of_Birth = 2021
GROUP BY First_Name
ORDER BY Total_Count DESC

ładowanie plików z magazynu obiektów w chmurze

Usługa Databricks zaleca używanie Auto Loader w potokach w przypadku większości zadań pozyskiwania danych z magazynu obiektów w chmurze lub z plików w woluminie Unity Catalog. Automatyczny ładowacz i potoki są przeznaczone do przyrostowego i idempotentnego ładowania stale rosnących danych w miarę ich napływu do magazynu w chmurze. Zobacz Co to jest moduł automatycznego ładowania? i Załaduj dane z magazynu obiektów.

Poniższy przykład odczytuje dane z magazynu w chmurze przy użyciu automatycznego modułu ładującego.

Python

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json")
  )

SQL

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
    format => "json"
  );

W poniższych przykładach użyto Auto Loader do tworzenia zestawów danych z plików CSV w Unity Catalog.

Python

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/Volumes/my_catalog/retail_org/customers/")
  )

SQL

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/retail_org/customers/",
  format => "csv"
)

Uwaga / Notatka

Zaloguj się do chmury

Auto Loader używa lokalizacji zewnętrznych Unity Catalog do uwierzytelniania w zasobach magazynu w chmurze. Musisz skonfigurować lokalizację zewnętrzną dla ścieżki magazynu, z której chcesz odczytać, i przyznać READ FILES użytkownikowi uprawnienia.

Aby pobierać dane z Azure Data Lake Storage, skonfiguruj lokalizację zewnętrzną wspieraną przez poświadczenia dostępu do magazynu, które odwołują się do kontenera danych. Aby uzyskać więcej informacji, zobacz Nawiązywanie połączenia z magazynem obiektów w chmurze z użyciem Unity Catalog.

Ładowanie danych z magistrali komunikatów

Pipeline'y można skonfigurować do pozyskiwania danych z magistrali komunikatów. Usługa Databricks zaleca używanie tabel strumieniowych z ciągłym wykonywaniem i ulepszonym skalowaniem automatycznym w celu zapewnienia najbardziej wydajnego pozyskiwania na potrzeby ładowania niskich opóźnień z szyn komunikatów. Aby uzyskać więcej informacji, zobacz Optymalizacja wykorzystania klastra potoku Lakeflow za pomocą automatycznego skalowania.

Na przykład poniższy kod konfiguruje tabelę przesyłania strumieniowego w celu pozyskiwania danych z Kafki za pomocą funkcji read_kafka.

Python

from pyspark import pipelines as dp

@dp.table
def kafka_raw():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "kafka_server:9092")
      .option("subscribe", "topic1")
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_raw AS
  SELECT *
  FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'topic1'
  );

Pozyskiwanie danych z Google Pub/Sub

Poniższy przykład tworzy tabelę strumieniową, która odczytuje dane z tematu Google Pub/Sub przy użyciu funkcji read_pubsub.

Python

@dp.table
def pubsub_raw():
  auth_options = {
    "clientId": client_id,
    "clientEmail": client_email,
    "privateKey": private_key,
    "privateKeyId": private_key_id
  }
  return (
    spark.readStream
      .format("pubsub")
      .option("subscriptionId", "my-subscription")
      .option("topicId", "my-topic")
      .option("projectId", "my-project")
      .options(auth_options)
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
  subscriptionId => 'my-subscription',
  projectId => 'my-project',
  topicId => 'my-topic',
  clientEmail => secret('pubsub-scope', 'clientEmail'),
  clientId => secret('pubsub-scope', 'clientId'),
  privateKeyId => secret('pubsub-scope', 'privateKeyId'),
  privateKey => secret('pubsub-scope', 'privateKey')
);

Usługa Databricks zaleca używanie tajemnic podczas podawania opcji autoryzacji. Zobacz Konfigurowanie dostępu do usługi Pub/Sub dla wszystkich opcji uwierzytelniania.

Aby pobierać dane z innych źródeł magistrali komunikatów, zobacz:

Ładowanie danych z Azure Event Hubs

Azure Event Hubs to usługa przesyłania strumieniowego danych, która zapewnia interfejs zgodny z platformą Apache Kafka. Aby ładować komunikaty z usługi Azure Event Hubs, można użyć łącznika Kafka dla Structured Streaming, dołączonego do środowiska uruchomieniowego potoku. Aby dowiedzieć się więcej na temat ładowania i przetwarzania komunikatów z usługi Azure Event Hubs, zobacz Używanie usługi Azure Event Hubs jako źródła danych potoku.

Ładowanie danych z systemów zewnętrznych

Potoki umożliwiają ładowanie danych z dowolnego źródła danych obsługiwanego przez usługę Azure Databricks. Zobacz Łączenie ze źródłami danych i usługami zewnętrznymi. Możesz również załadować dane zewnętrzne za pomocą usługi Lakehouse Federation dla obsługiwanych źródeł danych . Ponieważ środowisko Lakehouse Federation wymaga Databricks Runtime w wersji 13.3 LTS lub nowszej, aby z niego korzystać, skonfiguruj potok do używania kanału w wersji zapoznawczej.

Niektóre źródła danych nie mają równoważnej obsługi języka SQL. Jeśli nie możesz użyć Lakehouse Federation z jednym z tych źródeł danych, możesz użyć języka Python do przetwarzania danych ze źródła. Pliki źródłowe języka Python i SQL można dodawać do tego samego potoku. Poniższy przykład deklaruje zmaterializowany widok, aby uzyskać dostęp do bieżącego stanu danych w zdalnej tabeli PostgreSQL.

import dp

@dp.table
def postgres_raw():
  return (
    spark.read
      .format("postgresql")
      .option("dbtable", table_name)
      .option("host", database_host_url)
      .option("port", 5432)
      .option("database", database_name)
      .option("user", username)
      .option("password", password)
      .load()
  )

Ładowanie małych lub statycznych zestawów danych z magazynu obiektów w chmurze

Małe lub statyczne zestawy danych można załadować przy użyciu składni ładowania platformy Apache Spark. Potoki danych obsługują wszystkie formaty plików obsługiwane przez Apache Spark na platformie Azure Databricks. Aby uzyskać pełną listę, zobacz Opcje formatowania danych.

W poniższych przykładach pokazano ładowanie kodu JSON w celu utworzenia tabeli.

Python

@dp.table
def clickstream_raw():
  return (spark.read.format("json").load("/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"))

SQL

CREATE OR REFRESH MATERIALIZED VIEW clickstream_raw
AS SELECT * FROM read_files(
  "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"
)

Uwaga / Notatka

Funkcja read_files SQL jest wspólna dla wszystkich środowisk SQL w usłudze Azure Databricks. Jest to zalecany wzorzec bezpośredniego dostępu do plików przy użyciu języka SQL w potokach. Aby uzyskać więcej informacji, zobacz opcje .

Ładowanie danych z niestandardowego źródła danych języka Python

Niestandardowe źródła danych języka Python umożliwiają ładowanie danych w formatach niestandardowych. Możesz napisać kod do odczytu i zapisu w określonym zewnętrznym źródle danych lub użyć istniejącego kodu Python do odczytywania danych z własnych systemów wewnętrznych. Aby uzyskać więcej informacji na temat tworzenia źródeł danych języka Python, zobacz Niestandardowe źródła danych PySpark.

Poniższy przykład rejestruje niestandardowe źródło danych z nazwą formatu my_custom_datasource i odczytuje z niego w trybie wsadowym i strumieniowym.

from pyspark import pipelines as dp

# Assume `my_custom_datasource` is a custom Python custom data
# source that supports both batch and streaming reads, and has
# been registered using `spark.dataSource.register`.

# This creates a materialized view
@dp.table(name = "read_from_batch")
def read_from_batch():
    return spark.read.format("my_custom_datasource").load()

# This creates a streaming table
@dp.table(name = "read_from_streaming")
def read_from_streaming():
    return spark.readStream.format("my_custom_datasource").load()

Skonfiguruj tabelę streamingową, aby ignorować zmiany w źródłowej tabeli streamingowej

Domyślnie tabele przesyłania strumieniowego wymagają źródeł przeznaczonych wyłącznie do dołączania. Jeśli źródłowa tabela strumieniowa wymaga aktualizacji lub usunięć (na przykład w przypadku przetwarzania związanego z RODO, takiego jak „prawo do bycia zapomnianym”), użyj flagi skipChangeCommits, aby zignorować te zmiany. Ta flaga działa tylko z funkcją spark.readStreamoption() i nie może być używana, gdy tabela strumieniowania źródłowego jest celem funkcji create_auto_cdc_flow(). Aby uzyskać więcej informacji, zobacz Obsługa zmian w źródłowych tabelach usługi Delta Lake.

@dp.table
def b():
   return spark.readStream.option("skipChangeCommits", "true").table("A")

Bezpiecznie uzyskaj dostęp do poświadczeń magazynu za pomocą tajnych wartości w potoku danych

Możesz użyć Azure Databricks do przechowywania tajnych danych, takich jak klucze dostępu lub hasła. Aby skonfigurować tajne dane w potoku, użyj właściwości Spark w konfiguracji klastra w ustawieniach potoku. Zobacz Konfigurowanie klasycznych obliczeń dla potoków.

W poniższym przykładzie użyto tajnego kodu do przechowywania klucza dostępu wymaganego do odczytu danych wejściowych z konta magazynu Azure Data Lake Storage używając Auto Loader. Możesz użyć tej samej metody do skonfigurowania dowolnego tajnego elementu wymaganego przez potok, na przykład kluczy AWS do uzyskania dostępu do S3 lub hasła do magazynu metadanych Apache Hive.

Aby dowiedzieć się więcej na temat pracy z usługą Azure Data Lake Storage, zobacz Connect to Azure Data Lake Storage and Blob Storage.

Uwaga / Notatka

Należy dodać prefiks spark.hadoop. do klucza konfiguracji spark_conf, który ustawia tajną wartość.

{
  "id": "43246596-a63f-11ec-b909-0242ac120002",
  "storage": "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path>",
  "clusters": [
    {
      "spark_conf": {
        "spark.hadoop.fs.azure.account.key.<storage-account-name>.dfs.core.windows.net": "{{secrets/<scope-name>/<secret-name>}}"
      },
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5,
        "mode": "ENHANCED"
      }
    }
  ],
  "development": true,
  "continuous": false,
  "libraries": [
    {
      "notebook": {
        "path": "/Users/user@databricks.com/Pipeline Notebooks/pipeline quickstart"
      }
    }
  ],
  "name": "pipeline quickstart using ADLS2"
}

W tym przykładzie kodu zastąp następujące wartości.

Placeholder Zamień na
<container-name> Nazwa kontenera konta magazynu Azure.
<storage-account-name> Nazwa konta magazynu usługi ADLS.
<path> Ścieżka dla danych wyjściowych i metadanych z potoku danych.
<scope-name> Nazwa zakresu tajnych danych Azure Databricks.
<secret-name> Nazwa klucza zawierającego klucz dostępu do konta magazynu Azure. 
from pyspark import pipelines as dp

json_path = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path-to-input-dataset>"
@dp.create_table(
  comment="Data ingested from an ADLS2 storage account."
)
def read_from_ADLS2():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load(json_path)
  )

W tym przykładzie kodu zastąp następujące wartości.

Placeholder Zamień na
<container-name> Nazwa kontenera konta magazynu Azure, który przechowuje dane wejściowe.
<storage-account-name> Nazwa konta magazynu usługi ADLS.
<path-to-input-dataset> Ścieżka do wejściowego zestawu danych.
  • Last updated on