Функция с табличным значением `event_log`

Статья
04/18/2024

Область применения: проверка помечены да Databricks SQL Databricks Runtime 13.3 LTS и выше

Возвращает журнал событий для материализованных представлений, потоковых таблиц и конвейеров DLT.

Дополнительные сведения о журнале событий Delta Live Tables.

Синтаксис

event_log( { TABLE ( table_name ) | pipeline_id } )

Аргументы

table_name: имя материализованного представления или потоковой таблицы. Имя не должно содержать временную спецификацию. Если имя не задано, текущий каталог и схема используются для определения идентификатора.
pipeline_id: строковый идентификатор конвейера Delta Live Tables.

Возвраты

id STRING NOT NULL: уникальный идентификатор записи журнала событий.
sequence STRING NOT NULL: объект JSON, содержащий метаданные для идентификации и упорядочивания событий.
origin STRING NOT NULL: объект JSON, содержащий метаданные для источника события, например поставщика облачных служб, региона user_idили.pipeline_id
timestamp TIMESTAMP NOT NULL: время записи события в формате UTC.
message STRING NOT NULL: читаемое человеком сообщение, описывающее событие.
level STRING NOT NULL: уровень ведения журнала, например, INFO, WARNERRORили METRICS.
maturity_level STRING NOT NULL: стабильность схемы события. Возможные значения:
- STABLE: схема стабильна и не изменится.
- NULL: схема стабильна и не изменится. Значение может быть NULL , если запись была создана до maturity_level добавления поля (выпуск 2022.37).
- EVOLVING: схема не стабильна и может измениться.
- DEPRECATED: схема устарела, и среда выполнения Delta Live Tables может перестать создавать это событие в любое время.
error STRING: если произошла ошибка, сведения об ошибке, описывающие ошибку.
details STRING NOT NULL: объект JSON, содержащий структурированные сведения о событии. Это основное поле, используемое для анализа событий.
event_type STRING NOT NULL: тип события.

Использование

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

> CREATE VIEW event_log_raw AS SELECT * FROM event_log(table(my_mv));
> GRANT SELECT ON VIEW event_log_raw TO `user@databricks.com`;

Примеры

Дополнительные примеры см. в запросе журнала событий.

-- View the events on a materialized view
> SELECT timestamp, message, details
  FROM event_log(table(my_mv))
  WHERE level in ('INFO', 'WARN', 'ERROR')
  ORDER BY timestamp;

timestamp, message, details
---------------------------
2023-08-12 01:03:05.000, 'Flow "my_mv" is STARTING.', '{"flow_progress":{"status":"STARTING"}}'

-- Create a temp view with the latest update to the table/pipeline
> CREATE OR REPLACE TEMP VIEW latest_update AS
  SELECT origin.update_id AS id FROM event_log('<pipeline-ID>')
  WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1;

-- Query lineage information
> SELECT
  details:flow_definition.output_dataset as output_dataset,
  details:flow_definition.input_datasets as input_dataset
FROM
  event_log('<pipeline-ID>'),
  latest_update
WHERE
  event_type = 'flow_definition' AND origin.update_id = latest_update.id;

output_dataset, input_dataset
-----------------------------
customers, null
sales_orders_raw, null
sales_orders_cleaned, ["customers", "sales_orders_raw"]
sales_order_in_la, ["sales_orders_cleaned"]

-- Query data quality expectation history for a streaming table
> WITH expectations_parsed AS (
    SELECT
      explode(
        from_json(
          details:flow_progress.data_quality.expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations
    FROM
      event_log(table(my_st)),
      latest_update
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = latest_update.id
  )
  SELECT
    row_expectations.dataset as dataset,
    row_expectations.name as expectation,
    SUM(row_expectations.passed_records) as passing_records,
    SUM(row_expectations.failed_records) as failing_records
  FROM expectations_parsed
  GROUP BY
    row_expectations.dataset,
    row_expectations.name;

dataset, expectation, passing_records, failing_records
------------------------------------------------------
sales_orders_cleaned, valid_order_number, 4083, 0

Функция с табличным значением event_log