Запустите первую структурированную потоковую задачу

В этой статье приведены примеры кода и объяснение основных понятий, необходимых для выполнения первых структурированных запросов потоковой передачи в Azure Databricks. Структурированная потоковая передача можно использовать для практически в реальном времени и добавочных рабочих нагрузок обработки.

Структурированная потоковая передача — это одна из нескольких технологий, которые питают таблицы потоковой передачи в конвейерах Lakeflow. Databricks рекомендует использовать конвейеры Lakeflow для всех новых рабочих нагрузок ETL, приема и структурированной потоковой передачи. См. декларативные конвейеры Spark.

Использование автозагрузчика для чтения потоковых данных из хранилища объектов

В следующем примере демонстрируется загрузка данных JSON с помощью автозагрузчика, который используется cloudFiles для обозначения формата и параметров. Этот schemaLocation параметр включает вывод схемы и эволюцию. Вставьте следующий код в ячейку ноутбука Databricks и запустите ячейку, чтобы создать потоковый DataFrame с именем raw_df:

file_path = "/databricks-datasets/structured-streaming/events"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

raw_df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint_path)
    .load(file_path)
)

Как и другие операции чтения в Azure Databricks, настройка потокового чтения фактически не загружает данные. Перед началом потока необходимо активировать действие в данных.

Выполнение потокового преобразования

Структурированная потоковая передача поддерживает большинство преобразований, доступных в Azure Databricks и Spark SQL. Вы можете даже загружать модели MLflow как UDF и выполнять потоковые прогнозы в процессе преобразования.

В следующем примере кода выполняется простое преобразование для обогащения загруженных данных JSON дополнительной информацией с использованием функций Spark SQL.

from pyspark.sql.functions import col, current_timestamp

transformed_df = (raw_df.select(
    "*",
    col("_metadata.file_path").alias("source_file"),
    current_timestamp().alias("processing_time")
    )
)

В итоге transformed_df содержатся инструкции по загрузке и преобразованию каждой записи по мере их поступления в источник данных.

Выполните инкрементальную пакетную запись в Delta Lake

В следующем примере выполняется запись в Delta Lake, используя указанный путь к файлу и контрольную точку.

Параметр availableNow триггера указывает структурированной потоковой передаче обработать все ранее пропущенные записи из исходного набора данных и затем завершить работу, чтобы можно было безопасно выполнить следующий код, не опасаясь, что поток останется запущенным.

target_path = "/tmp/ss-tutorial/"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

transformed_df.writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .option("path", target_path)
    .start()

В этом примере новые записи не поступают в наш источник данных, поэтому повторное выполнение этого кода не включает новые записи.

Чтение данных из Delta Lake, преобразование и запись в Delta Lake

Delta Lake имеет обширную поддержку работы со структурированной потоковой передачей как источником, так и приемником. См. потоковое чтение и запись таблиц Delta Lake.

В следующем примере показан пример синтаксиса для добавочной загрузки всех новых записей из таблицы Delta Lake, объединения их с моментальным снимком другой таблицы Delta Lake и их записи в таблицу Delta Lake:

(spark.readStream
    .table("<table-name1>")
    .join(spark.read.table("<table-name2>"), on="<id>", how="left")
    .writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", "<checkpoint-path>")
    .toTable("<table-name3>")
)

У вас должны быть соответствующие разрешения, которые позволяют читать исходные таблицы, записывать в целевые таблицы и работать с указанным расположением контрольной точки. Заполните все параметры, обозначаемые угловыми скобками (<>) с помощью соответствующих значений для источников данных и приемников.

Чтение данных из Kafka, преобразование и запись в Kafka

Apache Kafka и другие системы обмена сообщениями предоставляют одну из самых низких задержек, доступных для больших наборов данных. Azure Databricks можно использовать для применения преобразований к данным, полученным из Kafka, а затем записывать данные обратно в Kafka.

В следующем примере кода демонстрируется простой шаблон для обогащения данных из Kafka путем объединения данных в таблице Delta Lake и последующей записи обратно в Kafka:

(spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("subscribe", "<topic>")
    .option("startingOffsets", "latest")
    .load()
    .join(spark.read.table("<table-name>"), on="<id>", how="left")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("topic", "<topic>")
    .option("checkpointLocation", "<checkpoint-path>")
    .start()
)

У вас должны быть соответствующие разрешения, настроенные для доступа к службе Kafka. Заполните все параметры, обозначаемые угловыми скобками (<>) с помощью соответствующих значений для источников данных и приемников. См. статью "Подключение к Apache Kafka".