Wczytywanie danych za pomocą notebooków

Notebooki w usłudze Azure Databricks zapewniają elastyczne, oparte na kodzie podejście do pobierania danych z różnych źródeł. Gdy narzędzia graficzne nie spełniają wymagań lub gdy potrzebujesz niestandardowej logiki do przekształcania danych, notebooki zapewniają pełną kontrolę nad procesem przetwarzania. Za pomocą języka Python (PySpark), SQL, Scala lub R można łączyć się ze źródłami danych, stosować przekształcenia i ładować dane do tabel katalogu Unity Catalog.

W tej jednostce dowiesz się, jak pobierać dane przy użyciu notebooków zarówno w scenariuszach wsadowych, jak i przesyłania strumieniowego.

Wczytywanie danych wsadowych za pomocą DataFrames

DataFrames są podstawą przetwarzania danych wsadowych w usłudze Databricks. Ramka danych to rozproszona kolekcja danych uporządkowana w nazwane kolumny, podobnie jak tabela w relacyjnej bazie danych. Używasz interfejsu spark.read API do ładowania danych z różnych źródeł do DataFrame.

Poniższy przykład odczytuje plik CSV do ramki danych:

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

Metoda format() określa typ pliku, podczas gdy option() metody konfigurują sposób analizowania danych. Opcja header informuje platformę Spark o użyciu pierwszego wiersza jako nazw kolumn i inferSchema automatycznie wykrywa typy danych.

Odczytywanie z różnych formatów plików

Usługa Azure Databricks obsługuje wiele formatów plików. Format i opcje można dostosować na podstawie źródła danych:

# Read JSON files
df_json = spark.read.format("json").load("/path/to/data.json")

# Read Parquet files (columnar format, efficient for analytics)
df_parquet = spark.read.format("parquet").load("/path/to/data.parquet")

# Read XML files (requires rowTag option)
df_xml = (spark.read
    .format("xml")
    .option("rowTag", "record")
    .load("/path/to/data.xml"))

<data>
  <record><id>1</id><name>Alice</name></record>
  <record><id>2</id><name>Bob</name></record>
</data>

Pliki Parquet przechowują dane w formacie kolumnowym, co zwiększa wydajność zapytań dla obciążeń analitycznych. Pliki JSON obsługują zagnieżdżone struktury i są typowe dla odpowiedzi interfejsu API. Pliki XML wymagają opcji rowTag do określenia, który element jest mapowany na każdy wiersz.

Nawiązywanie połączenia z zewnętrznymi bazami danych

W przypadku danych przechowywanych w relacyjnych bazach danych użyj połączeń JDBC do odczytu bezpośrednio ze źródła:

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

Przechowuj poświadczenia w sekretach Databricks, zamiast osadzać je na stałe w notesach. Zapytania można również wypychać do bazy danych przy użyciu podzapytania dbtable w opcji:

pushdown_query = "(SELECT id, name, amount FROM orders WHERE order_date > '2024-01-01') AS filtered_orders"
df = spark.read.format("jdbc").option("dbtable", pushdown_query).load()

Zapisywanie danych do tabel Unity Catalog

Po załadowaniu i przekształceniu danych zapisz je w tabeli Unity Catalog. Usługa Azure Databricks domyślnie używa formatu usługi Delta Lake, który zapewnia transakcje ACID i przechowywanie wersji:

df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

Opcja mode() określa sposób obsługi istniejących danych:

• nadpisanie: całkowicie zastępuje zawartość tabeli
• append: dodaje wiersze do istniejącej tabeli
• ignoruj: nie robi nic, jeśli tabela już istnieje
• błąd: Zgłasza wyjątek, jeśli tabela istnieje (wartość domyślna)

Pozyskiwanie danych przesyłanych strumieniowo za pomocą Strukturalizowanego Strumieniowania

Przesyłanie strumieniowe ze strukturą umożliwia ciągłe przetwarzanie danych w miarę ich napływu. Zamiast spark.read użyj spark.readStream, aby utworzyć obiekt DataFrame przesyłania strumieniowego, który przetwarza nowe dane przyrostowo.

Poniższy przykład odczytuje dane przesyłane strumieniowo z platformy Kafka:

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic-name")
    .option("startingOffsets", "latest")
    .load())

Opcja startingOffsets określa, gdzie rozpocząć odczytywanie podczas uruchamiania nowego zapytania. Użyj latest tylko dla nowych danych lub earliest do odczytywania od początku tematu. Po skonfigurowaniu lokalizacji punktu kontrolnego w strumieniu zapisu kolejne uruchomienia zostaną automatycznie wznowione z ostatniego przetworzonego przesunięcia, ignorując startingOffsets.

Przetwarzanie strumieni opartych na plikach

Możesz również przesyłać strumień ze źródeł plików, gdy nowe pliki pojawiają się w folderze.

df_file_stream = (spark.readStream
    .format("csv")
    .option("header", "true")
    .schema(defined_schema)
    .load("/Volumes/catalog/schema/volume/incoming/"))

W przypadku źródeł plików przesyłanych strumieniowo należy jawnie zdefiniować schemat zamiast używać polecenia inferSchema. Wnioskowanie schematu nie jest obsługiwane w przypadku źródeł przesyłania strumieniowego, ponieważ wymaga odczytania całego zestawu danych.

Zapisywanie danych przesyłanych strumieniowo do tabel

Służy writeStream do utrwalenia danych strumieniowych w tabelach wykazu Unity Catalog.

(df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.streaming_table"))

Tryb danych wyjściowych określa, które rekordy są zapisywane:

• Append: tylko nowe wiersze są zapisywane (domyślnie dla większości operacji)
• update: Zmienione wiersze są zapisywane
• complete: Wszystkie wiersze są przepisane (używane z agregacjami)

Lokalizacja punktu kontrolnego przechowuje informacje o postępie, co umożliwia wznowienie strumienia z miejsca, w którym zostało zatrzymane, jeśli zostanie przerwane.

Obsługa częściowo ustrukturyzowanych danych

Podczas pracy z częściowo ustrukturyzowanymi danymi, takimi jak JSON, typ danych VARIANT zapewnia elastyczność przechowywania i wykonywania zapytań dotyczących zagnieżdżonych struktur:

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))

Następnie można wykonywać zapytania dotyczące określonych pól przy użyciu variant_get funkcji lub składni SQL:

df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

Takie podejście działa dobrze, gdy schemat JSON różni się między rekordami lub zmienia się wraz z upływem czasu.