Pozyskiwanie danych za pomocą Spark Structured Streaming

Pozyskiwanie danych w czasie rzeczywistym umożliwia przetwarzanie zdarzeń w miarę ich występowania, a nie oczekiwanie na interwały wsadowe. Gdy twoja organizacja musi przechwytywać odczyty czujników IoT, dane strumienia kliknięć lub transakcje finansowe w miarę ich wystąpienia, Spark Structured Streaming stanowi podstawę ciągłego pozyskiwania danych do lakehouse.

W tej jednostce nauczysz się, jak skonfigurować zadania strumieniowania zorganizowanego, które odczytują ze źródeł strumieniowych i zapisują do tabel w Unity Catalog.

Zrozumieć model Structured Streaming

Przesyłanie strumieniowe ze strukturą traktuje dane przychodzące jako niezwiązaną tabelę, która stale rośnie wraz z nadejściem nowych danych. Używasz tego samego interfejsu API ramki danych, który już znasz, ale zamiast spark.read, używasz spark.readStream do stworzenia strumieniowej ramki danych, która przetwarza dane przyrostowo.

Rozważmy scenariusz, w którym dane czujników docierają stale z sprzętu produkcyjnego. Zamiast planować zadania wsadowe godzinowe w celu pozyskiwania tych danych, można skonfigurować zadanie przesyłania strumieniowego, które przetwarza poszczególne odczyty w ciągu kilku sekund od ich przybycia. Aparat przesyłania strumieniowego śledzi, które rekordy zostały przetworzone przy użyciu punktów kontrolnych, więc jeśli zadanie zostanie uruchomione ponownie, wznowi działanie bez ponownego przetwarzania danych ani pominięć.

Podstawowy wzorzec pozyskiwania danych przesyłanych strumieniowo obejmuje trzy kroki:

1. Konfigurowanie źródła przy użyciu spark.readStream ze szczegółami połączenia
2. Stosowanie przekształceń w celu kształtowania danych zgodnie z potrzebami
3. Zapisz do odbiornika przy użyciu writeStream lokalizacji punktu kontrolnego

Nawiązywanie połączenia z platformą Kafka i usługą Event Hubs

Usługa Azure Event Hubs udostępnia punkt końcowy zgodny z platformą Apache Kafka, co oznacza, że można użyć tego samego łącznika platformy Kafka dla obu usług. To ujednolicone podejście upraszcza kod podczas pracy z różnymi magistralami komunikatów.

Poniższy przykład odczytuje komunikaty z tematu platformy Kafka:

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

W przypadku usługi Azure Event Hubs nawiąż połączenie za pośrednictwem interfejsu zgodnego z platformą Kafka. Kluczową różnicą jest konfiguracja uwierzytelniania, która używa SASL z łańcuchem połączenia usługi Event Hubs.

eh_namespace = "your-namespace"
eh_name = "your-eventhub"
connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

kafka_options = {
    "kafka.bootstrap.servers": f"{eh_namespace}.servicebus.windows.net:9093",
    "subscribe": eh_name,
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{connection_string}";'
}

df_stream = (spark.readStream
    .format("kafka")
    .options(**kafka_options)
    .load())

Podczas odczytywania z platformy Kafka lub usługi Event Hubs za pomocą interfejsu zgodnego z platformą Kafka otrzymujesz standardowy schemat z kolumnami key, , valuetopicpartition, , offseti .timestamp Kolumny key i value zawierają dane binarne, więc zazwyczaj rzutujesz je na ciąg i analizujesz zawartość:

from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("device_id", StringType()) \
    .add("temperature", DoubleType()) \
    .add("timestamp", StringType())

parsed_df = (df_stream
    .select(col("value").cast("string").alias("json_value"))
    .select(from_json(col("json_value"), schema).alias("data"))
    .select("data.*"))

Korzystanie z natywnego łącznika usługi Event Hubs

Usługa Azure Databricks udostępnia również natywny łącznik usługi Event Hubs, który upraszcza uwierzytelnianie i zapewnia bezpośredni dostęp do funkcji specyficznych dla usługi Event Hubs. Ten łącznik używa eventhubs formatu:

connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .load())

Natywny łącznik usługi Event Hubs zwraca inny schemat niż interfejs zgodny z platformą Kafka. Ładunek zdarzenia jest przechowywany w body kolumnie (zamiast value), wraz z dodatkowymi kolumnami metadanych specyficznych dla usługi Event Hubs:

# Extract and parse the event payload
payload = events.selectExpr("CAST(body AS STRING) AS payload")

Możesz skonfigurować dodatkowe opcje usługi Event Hubs, takie jak grupa odbiorców i pozycja początkowa:

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .option("eventhubs.consumerGroup", "$Default")
    .option("eventhubs.startingPosition", '{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}')
    .load())

Opcja startingPosition określa, gdzie rozpocząć odczytywanie zdarzeń. Użyj -1 polecenia dla najnowszych zdarzeń lub skonfiguruj określone przesunięcie, numer sekwencji lub czas w kolejce na potrzeby pozyskiwania.

Konfigurowanie lokalizacji punktów kontrolnych

Punkty kontrolne to mechanizm, który sprawia, że Strukturalne Przesyłanie Strumieniowe jest odporne na błędy. Katalog punktów kontrolnych przechowuje informacje o przesunięciach, dokładnie śledząc, które rekordy zostały przetworzone. Jeśli zadanie przesyłania strumieniowego zostanie nieoczekiwanie zatrzymane, może zostać uruchomione ponownie i kontynuować z ostatniego zatwierdzonego punktu kontrolnego.

Po uruchomieniu zapisu strumieniowego określ lokalizację punktu kontrolnego:

checkpoint_path = "/checkpoints/sensor-ingestion"

query = (parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .toTable("catalog.schema.sensor_readings"))

Przechowywanie punktów kontrolnych w niezawodnej, trwałej lokalizacji, takiej jak Usługa Azure Data Lake Storage. Pliki punktu kontrolnego są małe, ale krytyczne dla odzyskiwania danych, dlatego należy skonfigurować lokalizację magazynu przy użyciu odpowiednich zasad tworzenia kopii zapasowych i przechowywania.

Zapisywanie danych przesyłanych strumieniowo do tabel Unity Catalog

Pisanie do tabel Unity Catalog używa formatu Delta Lake, który obsługuje zarówno zapisy strumieniowe, jak i transakcje ACID. API writeStream odzwierciedla interfejs wsadowy API write, ale zawiera opcje przetwarzania ciągłego.

(parsed_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("bronze.iot.sensor_readings"))

Tryb danych wyjściowych określa sposób zapisywania wyników:

• append: do tabeli są dodawane nowe wiersze. Używaj tego trybu dla większości zapytań przetwarzania.
• update: Zmienione wiersze są aktualizowane. Użyj z agregacjami, które generują aktualizacje.
• complete: wszystkie wiersze są ponownie napisane. Używaj z agregacjami, które wymagają pełnego ponownego obliczenia.

W przypadku scenariuszy przetwarzania, w których dołączasz nieprzetworzone dane, domyślny tryb append jest odpowiedni.

Sterowanie przetwarzaniem za pomocą wyzwalaczy

Wyzwalacze określają, kiedy i jak często aparat przesyłania strumieniowego przetwarza dane. Usługa Azure Databricks obsługuje kilka opcji wyzwalacza:

W przypadku ciągłego pozyskiwania, które jest uruchamiane na czas nieokreślony, pomiń opcję wyzwalacza lub użyj interwału czasu przetwarzania. W przypadku przyrostowych wzorców wsadowych, w których chcesz przetwarzać wszystkie dostępne dane zgodnie z harmonogramem, użyj polecenia availableNow:

(parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .trigger(availableNow=True)
    .toTable("catalog.schema.table_name"))

Wyzwalacz availableNow przetwarza wszystkie nieprzetworzone dane od ostatniego punktu kontrolnego, a następnie kończy działanie. Ten wzorzec można zaplanować przy użyciu zadań usługi Databricks do uruchamiania w regularnych odstępach czasu, łącząc wydajność przesyłania strumieniowego (tylko przetwarzanie nowych danych) z przewidywalnością zaplanowanych zadań wsadowych.

Monitorowanie zapytań przesyłania strumieniowego

Aktywne zapytania przesyłania strumieniowego są wyświetlane w interfejsie użytkownika platformy Spark na karcie Przesyłanie strumieniowe ze strukturą . Możesz również programowo uzyskać dostęp do stanu zapytania:

query = parsed_df.writeStream.format("delta").option("checkpointLocation", checkpoint_path).toTable("my_table")

# Check if the query is still running
print(query.isActive)

# Get the current status
print(query.status)

# View recent progress
print(query.recentProgress)

Użyj metody awaitTermination(), aby zablokować wykonanie do momentu, gdy strumień zakończy działanie lub wystąpi jego niepowodzenie. W przypadku obciążeń produkcyjnych zaimplementuj obsługę błędów w celu przechwytywania i reagowania na błędy przesyłania strumieniowego:

try:
    query.awaitTermination()
except Exception as e:
    print(f"Stream failed: {e}")

W środowiskach produkcyjnych skonfiguruj monitorowanie i alerty, aby wykrywać, kiedy zadania przesyłania strumieniowego opóźniają się lub napotykają błędy. Metryki przesyłania strumieniowego zawierają informacje o opóźnieniach, które pomagają zrozumieć, czy przetwarzanie nadąża za przychodzącymi danymi.