Lekérdezési előzmények rendszertáblázatának referenciája

Ez a cikk a lekérdezéselőzmények rendszertáblájára vonatkozó információkat tartalmazza, beleértve a tábla sémájának vázlatát is.

tábla elérési útja: Ez a rendszertábla található a system.query.history.

A lekérdezéselőzmények táblájának használata

A lekérdezéselőzmények táblázata sql-raktárak vagy kiszolgáló nélküli számítás használatával futtatott lekérdezések rekordjait tartalmazza jegyzetfüzetekhez és feladatokhoz. A tábla az összes munkaterület fiókszintű rekordjait tartalmazza ugyanabban a régióban, ahonnan a táblához hozzáfér.

Alapértelmezés szerint csak a rendszergazdák férhetnek hozzá a rendszertáblához. Ha meg szeretné osztani a tábla adatait egy felhasználóval vagy csoporttal, a Databricks azt javasolja, hogy minden felhasználóhoz vagy csoporthoz hozzon létre egy dinamikus nézetet. Lásd: Dinamikus nézet létrehozása.

Lekérdezési előzmények rendszertáblázatának sémája

A lekérdezéselőzmény-tábla a következő sémát használja:

Rekord lekérdezésprofiljának megtekintése

Ha egy lekérdezés lekérdezési profiljára szeretne navigálni a lekérdezéselőzménytáblában lévő rekord alapján, tegye a következőket:

1. Azonosítsa a rekordot, majd másolja ki a rekordot statement_id.
2. A rekordra workspace_id hivatkozva győződjön meg arról, hogy ugyanabba a munkaterületre van bejelentkezve, mint a rekord.
3. Kattintson a Lekérdezési előzmények elemre a munkaterület oldalsávjában.
4. Az Utasítás azonosítója mezőbe illessze be a statement_id rekordot.
5. Kattintson egy lekérdezés nevére. Megjelenik a lekérdezési metrikák áttekintése.
6. Kattintson a Lekérdezésprofil megtekintése elemre.

A lekérdezési előzmények materializálása a metaadattárból

Az alábbi kód egy óránként, naponta vagy hetente futó feladat létrehozására használható, amely egy metaadattár lekérdezési előzményeit hozza létre. Ennek megfelelően állítsa be a HISTORY_TABLE_PATH és LOOKUP_PERIOD_DAYS változókat.

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