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

Операции DataFrame или функции табличного значения SQL можно использовать для запроса данных о состоянии и метаданных структурированной потоковой передачи. Используйте эти функции для отслеживания сведений о состоянии для запросов с отслеживанием состояния структурированной потоковой передачи, которые могут быть полезны для мониторинга и отладки.

Для запроса данных о состоянии или метаданных необходимо иметь доступ на чтение к пути контрольной точки для потокового запроса. Функции, описанные в этой статье, предоставляют доступ только для чтения к данным состояния и метаданным. Для запроса сведений о состоянии можно использовать только пакетную семантику чтения.

Требования

• Используйте одну из следующих конфигураций вычислений:
  • Databricks Runtime 16.3 и более поздних версий на вычислительных системах, настроенных в стандартном режиме доступа.
  • Databricks Runtime 14.3 LTS и выше на вычислительных системах, настроенных на режим выделенного доступа или без режима изоляции.
• Доступ на чтение к пути контрольной точки, используемому потоковым запросом.

Чтение хранилища состояний структурированной потоковой передачи

Вы можете считывать сведения о хранилище состояний для запросов структурированной потоковой передачи, выполняемых в любой поддерживаемой среде выполнения Databricks. Используйте следующий синтаксис:

Питон

df = (spark.read
  .format("statestore")
  .load("/checkpoint/path"))

Scala

val df = spark.read
  .format("statestore")
  .load("/checkpoint/path")

SQL

SELECT * FROM read_statestore('/checkpoint/path')

Параметры и схема API для чтения состояния

Полный список параметров формата см. в statestore".

Выходные данные имеют следующую схему:

В Databricks Runtime 16.4 LTS и более поздних версиях, если для параметра readChangeFeed задано значение true, выходные данные имеют следующую схему:

См. табличную функцию read_statestore.

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

Доступно в Databricks Runtime 16.4 LTS и более поздних версиях. Чтобы узнать, как состояние изменяется по микропакетам, а не просматривать полное состояние для одного микропакета, установите для readChangeFeed значение true и укажите changeStartBatchId. При необходимости укажите changeEndBatchId. Полный список параметров см. в разделе "Хранилище состояний".

Например, чтобы читать изменения состояния начиная с пакета 2 и до последнего зафиксированного пакета:

Питон

df = (spark.read
  .format("statestore")
  .option("readChangeFeed", True)
  .option("changeStartBatchId", 2)
  .load("<checkpointLocation>")
)

Scala

val df = spark.read
  .format("statestore")
  .option("readChangeFeed", true)
  .option("changeStartBatchId", 2)
  .load("<checkpointLocation>")

SQL

SELECT * FROM read_statestore(
    '<checkpointLocation>',
    readChangeFeed => true,
    changeStartBatchId => 2
);

Схема вывода включает дополнительные batch_id и change_type столбцы. Полный вариант схемы см. в разделе "Параметры и схема средства чтения состояния".

Чтение метаданных состояния структурированной потоковой передачи

Доступно в Databricks Runtime 14.3 LTS или более поздней версии. Вы можете считывать сведения о метаданных состояния для запросов структурированной потоковой передачи:

Питон

df = (spark.read
  .format("state-metadata")
  .load("<checkpointLocation>"))

Scala

val df = spark.read
  .format("state-metadata")
  .load("<checkpointLocation>")

SQL

SELECT * FROM read_state_metadata('/checkpoint/path')

Возвращенные данные имеют следующую схему:

См. табличную функцию read_state_metadata.

Пример: Запрос одной стороны соединения двух потоков

Используйте следующий синтаксис, чтобы запросить левую сторону stream-stream соединения.

Питон

left_df = (spark.read
  .format("statestore")
  .option("joinSide", "left")
  .load("/checkpoint/path"))

Scala

val leftDf = spark.read
  .format("statestore")
  .option("joinSide", "left")
  .load("/checkpoint/path")

SQL

SELECT * FROM read_statestore(
    '/checkpoint/path',
    joinSide => 'left'
);

Пример: Запрос к хранилищу состояний для потока с несколькими операторами состояния

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

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

Питон

df = (spark.read
  .format("state-metadata")
  .load("<checkpointLocation>"))

Scala

val df = spark.read
  .format("state-metadata")
  .load("<checkpointLocation>")

SQL

SELECT * FROM read_state_metadata('/checkpoint/path')

В следующей таблице представлен пример выходных данных метаданных хранилища состояний:

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

Питон

left_df = (spark.read
  .format("statestore")
  .option("operatorId", 1)
  .load("/checkpoint/path"))

Scala

val leftDf = spark.read
  .format("statestore")
  .option("operatorId", 1)
  .load("/checkpoint/path")

SQL

SELECT * FROM read_statestore(
    '/checkpoint/path',
    operatorId => 1
);