查询历史记录系统表参考

本文包含有关查询历史记录系统表的信息，包括表架构的大纲。

表路径：此系统表位于 system.query.history.

使用查询历史记录表

查询历史记录表包含使用 SQL 仓库或无服务器计算在笔记本和作业中运行的查询记录。 该表包括来自访问表的同一区域中所有工作区的帐户范围记录。

默认情况下，只有管理员有权访问系统表。 如果要与用户或组共享表的数据，Databricks 建议为每个用户或组创建动态视图。 请参阅创建动态视图。

查询历史记录系统表架构

查询历史记录表使用以下架构：

查看记录的查询配置文件

若要根据查询历史表中的记录导航到查询配置文件，请执行以下操作：

1. 确定感兴趣的记录，然后复制记录的statement_id。
2. 请参考记录的 workspace_id 以确保您登录的工作区与记录相同。
3. 单击工作区边栏中的查询历史记录。
4. 在“语句 ID”字段中，将 statement_id 粘贴到记录上。
5. 单击查询的名称。 此时将显示查询指标的概述。
6. 单击“查看查询配置文件”。

了解查询来源列

该 query_source 列包含语句执行中涉及的 Azure Databricks 实体的一组唯一标识符。

query_source如果该列包含多个 ID，则表示语句执行是由多个实体触发的。 例如，作业结果可能会触发调用 SQL 查询的警报。 在此示例中，将填充所有三个 ID 到 query_source。 此列的值不按执行顺序排序。

可能的查询源包括：

• alert_id：从警报触发的语句
• sql_query_id：在此 SQL 编辑器 会话中执行的语句
• dashboard_id：从仪表板执行的语句
• legacy_dashboard_id：在旧仪表板中执行的语句
• genie_space_id：从 Genie 空间执行的语句
• notebook_id：从笔记本执行的语句
• job_info.job_id：在作业中执行的语句
• job_info.job_run_id：从作业运行执行的语句
• job_info.job_task_run_id：在作业任务运行中执行的语句

query_source的有效组合

以下示例显示如何根据查询的运行方式填充query_source列：

• 在作业运行期间执行的查询包括填充 job_info 的结构体：

  {
alert_id: null,
sql_query_id: null,
dashboard_id: null,
notebook_id: null,
job_info: {
job_id: 64361233243479,
job_run_id: null,
job_task_run_id: 110378410199121
},
legacy_dashboard_id: null,
genie_space_id: null
}

• 来自旧仪表板的查询包括一个sql_query_id 和 legacy_dashboard_id：

  {
alert_id: null,
sql_query_id: 7336ab80-1a3d-46d4-9c79-e27c45ce9a15,
dashboard_id: null,
notebook_id: null,
job_info: null,
legacy_dashboard_id: 1a735c96-4e9c-4370-8cd7-5814295d534c,
genie_space_id: null
}

• 来自警报的查询包括sql_query_id和alert_id：

  {
alert_id: e906c0c6-2bcc-473a-a5d7-f18b2aee6e34,
sql_query_id: 7336ab80-1a3d-46d4-9c79-e27c45ce9a15,
dashboard_id: null,
notebook_id: null,
job_info: null,
legacy_dashboard_id: null,
genie_space_id: null
}

• 来自仪表板的查询包括一个 dashboard_id，但不包含 job_info：

  {
alert_id: null,
sql_query_id: null,
dashboard_id: 887406461287882,
notebook_id: null,
job_info: null,
legacy_dashboard_id: null,
genie_space_id: null
}

具体化元存储中的查询历史记录

以下代码可用于创建每小时、每天或每周运行的作业，以具体化元存储中的查询历史记录。 相应地调整 HISTORY_TABLE_PATH 和 LOOKUP_PERIOD_DAYS 变量。

from delta.tables import *
from pyspark.sql.functions import *
from pyspark.sql.types import *

HISTORY_TABLE_PATH = "jacek.default.history"
# Adjust the lookup period according to your job schedule
LOOKUP_PERIOD_DAYS = 1

def table_exists(table_name):
    try:
        spark.sql(f"describe table {table_name}")
        return True
    except Exception:
        return False

def save_as_table(table_path, df, schema, pk_columns):
    deltaTable = (
        DeltaTable.createIfNotExists(spark)
        .tableName(table_path)
        .addColumns(schema)
        .execute()
    )

    merge_statement = " AND ".join([f"logs.{col}=newLogs.{col}" for col in pk_columns])

    result = (
        deltaTable.alias("logs")
        .merge(
            df.alias("newLogs"),
            f"{merge_statement}",
        )
        .whenNotMatchedInsertAll()
        .whenMatchedUpdateAll()
        .execute()
    )
    result.show()

def main():
    df = spark.read.table("system.query.history")
    if table_exists(HISTORY_TABLE_PATH):
        df = df.filter(f"update_time >= CURRENT_DATE() - INTERVAL {LOOKUP_PERIOD_DAYS} days")
    else:
        print(f"Table {HISTORY_TABLE_PATH} does not exist. Proceeding to copy the whole source table.")

    save_as_table(
        HISTORY_TABLE_PATH,
        df,
        df.schema,
        ["workspace_id", "statement_id"]
    )

main()