Подключение к Apache Kafka

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

Дополнительные сведения о Kafka см. в документации по Apache Kafka.

Чтение данных из Kafka

kafka Используйте формат для настройки подключений к Kafka. Ниже приведен пример потокового чтения:

Python

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()
)

Scala

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()

SQL

CREATE OR REFRESH STREAMING TABLE <table_name> AS
SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<server:ip>',
  subscribe => '<topic>'
);

Azure Databricks также поддерживает пакетные операции чтения из Kafka, как показано в следующем примере:

Python

df = (spark.read
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
)

Scala

val df = spark.read
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()

SQL

SELECT * FROM read_kafka(
  bootstrapServers => '<server:ip>',
  subscribe => '<topic>',
  startingOffsets => 'earliest',
  endingOffsets => 'latest'
);

Для добавочной пакетной загрузки Databricks рекомендует использовать Kafka с Trigger.AvailableNow. См.: AvailableNow инкрементную пакетную обработку.

В Databricks Runtime 13.3 LTS и более поздних версиях Azure Databricks также предоставляет функцию SQL для чтения данных Kafka. Потоковая передача с помощью SQL поддерживается только в конвейерах Lakeflow или с таблицами потоковой передачи в Databricks SQL. См. табличную функцию read_kafka.

Настройка средства чтения структурированной потоковой передачи Kafka

Для пакетных и потоковых запросов необходимо задать серверы начальной загрузки для источника Kafka следующим образом:

Чтобы задать разделы подписки, необходимо указать один из следующих параметров:

Полный список доступных параметров см. в Kafka .

Схема строк Kafka

Средство чтения структурированной потоковой передачи Kafka возвращает строки со следующей схемой:

key и value всегда десериализуются как массивы байтов с ByteArrayDeserializer. Используйте операции DataFrame (например cast("string") , или from_avro) для явной десериализации ключей и значений.

Запись данных в Kafka

Ниже приведен пример потоковой записи в Kafka:

Python

(df.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .start()
)

Scala

df.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .start()

Azure Databricks также поддерживает семантику пакетной записи в приемники данных Kafka, как показано в следующем примере:

Python

(df.write
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .save()
)

Scala

df.write
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .save()

Настройка модуля записи структурированной потоковой передачи Kafka

Ниже приведены распространенные варианты записи в Kafka:

Полный список доступных параметров см. в разделе Kafka sink.

Схема модуля записи Kafka

При записи данных в Kafka указанный кадр данных может содержать следующие поля:

Проверка подлинности

Azure Databricks поддерживает несколько методов проверки подлинности для Kafka, включая учетные данные службы каталога Unity, SASL/SSL и облачные варианты для AWS MSK, Центры событий Azure и Google Cloud Managed Kafka. См. раздел Аутентификация.

Получение метрик Kafka

Чтобы отслеживать отставание по Kafka для потокового запроса, используйте метрики avgOffsetsBehindLatest, maxOffsetsBehindLatest и minOffsetsBehindLatest. Эти метрики сообщают о средней, максимальной и минимальной задержке смещения по всем секциям разделов, подписанных по сравнению с последними смещениями в Kafka. См. статью Чтение метрик в интерактивном режиме.

Чтобы оценить оставшиеся данные для чтения запроса, используйте метрику estimatedTotalBytesBehindLatest . Эта метрика оценивает общее количество байтов, оставшихся во всех подписанных секциях на основе пакетов, обработанных за последние 300 секунд. Вы можете изменить период времени, используемый для этой оценки, задав bytesEstimateWindowLength параметр.

Например, чтобы задать длину окна в 10 минут:

Python

df = (spark.readStream
  .format("kafka")
  .option("bytesEstimateWindowLength", "10m") # m for minutes, you can also use "600s" for 600 seconds
)

Scala

val df = spark.readStream
  .format("kafka")
  .option("bytesEstimateWindowLength", "10m") // m for minutes, you can also use "600s" for 600 seconds

Если вы выполняете поток в записной книжке, эти метрики можно просмотреть на вкладке "Необработанные данные " на панели мониторинга хода выполнения потокового запроса:

{
  "sources": [
    {
      "description": "KafkaV2[Subscribe[topic]]",
      "metrics": {
        "avgOffsetsBehindLatest": "4.0",
        "maxOffsetsBehindLatest": "4",
        "minOffsetsBehindLatest": "4",
        "estimatedTotalBytesBehindLatest": "80.0"
      }
    }
  ]
}

Дополнительные сведения см. в Мониторинг структурированных запросов потоковой передачи Azure Databricks.

Пример переноса данных из Kafka в Delta Lake

В следующем примере показан полный рабочий процесс для добавочной потоковой записи из Kafka в таблицу Delta Lake с помощью триггера availableNow . Этот подход можно использовать для добавочных рабочих нагрузок приема данных.

В этом примере используется фиксированная схема JSON. Для других форматов, таких как Avro или Protobuf, используйте from_avro или from_protobuf. Вы также можете интегрироваться с реестром схем. См . пример с реестром схем.

Python

from pyspark.sql.functions import from_json, col

# Define simple JSON schemas for key and value
key_schema = "user_id STRING"
value_schema = "event_type STRING, event_ts TIMESTAMP"

# Configure Kafka options with service credentials
kafka_options = {
  "kafka.bootstrap.servers": "<bootstrap-server>:9092",
  "subscribe": "<topic-name>",
  "databricks.serviceCredential": "<service-credential-name>",
}

# Read from Kafka and parse JSON
parsed_df = (spark.readStream
  .format("kafka")
  .options(**kafka_options)
  .load()
  .select(
    from_json(col("key").cast("string"), key_schema).alias("key"),
    from_json(col("value").cast("string"), value_schema).alias("value")
  )
  .select("key.*", "value.*")
)

# Write to Delta table
query = (parsed_df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .trigger(availableNow=True)
  .toTable("catalog.schema.events_table")
)

query.awaitTermination()

Scala

import org.apache.spark.sql.functions.{from_json, col}
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.StructType

// Define JSON schemas for key and value
val keySchema = "user_id STRING"
val valueSchema = "event_type STRING, event_ts TIMESTAMP"

// Configure Kafka options with service credentials
val kafkaOptions = Map(
  "kafka.bootstrap.servers" -> "<bootstrap-server>:9092",
  "subscribe" -> "<topic-name>",
  "databricks.serviceCredential" -> "<service-credential-name>"
)

// Read from Kafka and parse JSON
val parsedDF = spark.readStream
  .format("kafka")
  .options(kafkaOptions)
  .load()
  .select(
    from_json(col("key").cast("string"), StructType.fromDDL(keySchema)).alias("key"),
    from_json(col("value").cast("string"), StructType.fromDDL(valueSchema)).alias("value")
  )
  .select("key.*", "value.*")

// Write to Delta table
val query = parsedDF.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .toTable("catalog.schema.events_table")

query.awaitTermination()

SQL

-- Create a streaming table from Kafka using read_kafka
CREATE OR REFRESH STREAMING TABLE catalog.schema.events_table AS
SELECT
  key::string:user_id AS user_id,
  value::string:event_type AS event_type,
  to_timestamp(value::string:event_ts) AS event_ts
FROM STREAM read_kafka(
  bootstrapServers => '<bootstrap-server>:9092',
  subscribe => '<topic-name>',
  serviceCredential => '<service-credential-name>'
);