Referensi tabel sistem riwayat kueri

Artikel ini menyertakan informasi tentang tabel sistem riwayat kueri, termasuk kerangka skema tabel.

Jalur tabel: Tabel sistem ini terletak di system.query.history.

Menggunakan tabel riwayat kueri

Tabel riwayat kueri menyertakan rekaman untuk kueri yang dijalankan menggunakan gudang SQL atau komputasi tanpa server untuk buku catatan dan pekerjaan . Tabel menyertakan rekaman seluruh akun dari semua ruang kerja di wilayah yang sama tempat Anda mengakses tabel.

Secara default, hanya admin yang memiliki akses ke tabel sistem. Jika Anda ingin berbagi data tabel dengan pengguna atau grup, Databricks merekomendasikan untuk membuat tampilan dinamis untuk setiap pengguna atau grup. Lihat Membuat tampilan dinamis.

Skema tabel sistem riwayat kueri

Tabel riwayat kueri menggunakan skema berikut:

Menampilkan profil kueri untuk rekaman

Untuk menavigasi ke profil kueri berdasarkan rekaman dalam tabel riwayat kueri, lakukan hal berikut:

1. Identifikasi catatan yang menarik, lalu salin rekaman statement_id.
2. Referensikan rekaman workspace_id untuk memastikan Anda masuk ke ruang kerja yang sama dengan rekaman.
3. Klik Riwayat Kueri di bilah samping ruang kerja.
4. Di bidang ID Pernyataan, tempelkan statement_id pada rekaman.
5. Klik nama kueri. Gambaran umum metrik kueri muncul.
6. Klik Lihat profil kueri.

Mewujudkan riwayat kueri dari metastore Anda

Kode berikut dapat digunakan untuk membuat tugas yang dijalankan setiap jam, setiap hari, atau setiap minggu untuk mematerialisasikan riwayat kueri dari sebuah metastore. Sesuaikan variabel HISTORY_TABLE_PATH dan 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()