event_log 表值函数

适用于： Databricks SQL Databricks Runtime 13.3 LTS 及更高版本

返回 具体化视图、流式处理表和 Lakeflow 声明性管道的事件日志。

了解更多关于 Lakeflow 声明性管道 事件日志的信息。

语法

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：架构稳定，不会改变。 如果记录是在添加 NULL 字段之前（版本 2022.37）创建的，则该值可能是 maturity_level。
  • EVOLVING：架构不稳定，可能会发生变化。
  • DEPRECATED：架构已弃用，Lakeflow 声明性 Pipelines 运行时可以随时停止生成此事件。
• error STRING：如果出现错误，则为描述错误的详细信息。
• details STRING NOT NULL：一个 JSON 对象，其中包含事件的结构化详细信息。 这是用于分析事件的主要字段。
• event_type STRING NOT NULL：事件类型。

示例

有关更多示例，请参阅 Lakeflow 声明性管道事件日志。

-- 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

相关文章

• 监控 Lakeflow 的声明性管道
• Lakeflow 声明性管道事件日志
• “Lakeflow” 声明性管道
• 创建流表