event_log
表值函数
适用于: Databricks SQL Databricks Runtime 13.3 LTS 及更高版本
详细了解增量实时表的事件日志。
注意
event_log
表值函数只能由流式处理表或具体化视图的所有者调用,而通过 event_log
表值函数创建的视图只能由流式处理表或具体化视图的所有者查询。 无法与其他用户共享视图。
语法
event_log( { TABLE ( table_name ) | pipeline_id } )
参数
- table_name:具体化视图或流式处理表的名称。 名称不得包含时态规范。 如果名称未限定,则使用当前目录和架构来限定标识符。
pipeline_id
:增量实时表管道的字符串标识符。
返回
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
、WARN
、ERROR
或METRICS
。maturity_level STRING NOT NULL
:事件架构的稳定性。 可能的值包括:STABLE
:架构稳定,不会改变。NULL
:架构稳定,不会改变。 如果记录是在添加maturity_level
字段之前(版本 2022.37)创建的,则该值可能是NULL
。EVOLVING
:架构不稳定,可能会发生变化。DEPRECATED
:架构已弃用,Delta Live Tables 运行时可能随时停止生成此事件。
error STRING
:如果出现错误,则为描述错误的详细信息。details STRING NOT NULL
:一个 JSON 对象,其中包含事件的结构化详细信息。 这是用于分析事件的主要字段。event_type STRING NOT NULL
:事件类型。
示例
有关更多示例,请参阅查询事件日志。
-- 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