Referência do sistema de histórico de consultas table

Este artigo inclui informações sobre o sistema de histórico de consultas table, incluindo um esboço dos componentes schemado sistema table.

Table caminho: Este sistema table está localizado em system.query.history.

Usando o histórico de consultas table

O histórico de consultas table inclui registos para cada instrução SQL executada usando warehouses SQL. O table inclui registros de toda a conta de todos os espaços de trabalho na mesma região a partir da qual você acessa o table.

Por padrão, apenas os administradores têm acesso ao sistema table. Se você quiser compartilhar os dados do tablecom um usuário ou grupo, o Databricks recomenda a criação de uma exibição dinâmica para cada usuário ou grupo. Consulte Criar uma vista dinâmica.

Sistema de histórico de consultas tableschema

O histórico de consultas table usa os seguintes schema:

Exibir o perfil de consulta para um registro

Para navegar até o perfil de consulta de uma consulta com base em um registro no histórico de consultas table, faça o seguinte:

1. Identifique o registro de interesse e, em seguida, copie o statement_idarquivo .
2. Faça referência ao registro workspace_id para garantir que você esteja conectado ao mesmo espaço de trabalho que o registro.
3. Clique em Histórico de Consultas na barra lateral do espaço de trabalho.
4. No campo ID da declaração, cole o statement_id no registro.
5. Clique no nome de uma consulta. Uma visão geral das métricas de consulta é exibida.
6. Clique em Ver perfil de consulta.

Materializar o histórico de consultas a partir do seu metastore

O código a seguir pode ser usado para agendar uma tarefa para executar a cada hora, diariamente ou semanalmente, para materializar o histórico de consultas de um metastore. Ajuste as variáveis HISTORY_TABLE_PATH e LOOKUP_PERIOD_DAYS de acordo.

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