Sistem tablosu sorgu geçmişi referansı

Bu makale, sorgu geçmişi sistem tablosuyla ilgili bilgileri ve tablonun şemasının ana hattını içerir.

Tablo yolu: Bu sistem tablosu system.query.historykonumunda bulunur.

Sorgu geçmişi tablosunu kullanma

Sorgu geçmişi tablosu, SQL ambarları kullanılarak çalıştırılacak sorgular veya not defterleri için sunucusuz işlem ve işlerikayıtları içerir. Tablo, tabloya erişebileceğiniz aynı bölgedeki tüm çalışma alanlarındaki hesap genelindeki kayıtları içerir.

Varsayılan olarak, sistem tablosuna yalnızca yöneticiler erişebilir. Tablonun verilerini bir kullanıcı veya grupla paylaşmak isterseniz Databricks, her kullanıcı veya grup için dinamik bir görünüm oluşturmanızı önerir. Bkz. Dinamik görünüm oluşturma.

Sorgu geçmişi sistem tablosu şeması

Sorgu geçmişi tablosu aşağıdaki şemayı kullanır:

Kaydın sorgu profilini görüntüleme

Sorgu geçmişi tablosundaki bir kaydı temel alarak sorgunun sorgu profiline gitmek için aşağıdakileri yapın:

1. İlgi alanının kaydını belirleyin ve kaydın statement_idöğesini kopyalayın.
2. Kayıtla workspace_id aynı çalışma alanında oturum açtığınızdan emin olmak için kayıtlara başvurun.
3. Çalışma alanı kenar çubuğunda Sorgu Geçmişi'ne tıklayın .
4. Deyim Kimliği alanına, kaydına yapıştırınstatement_id.
5. Sorgunun adına tıklayın. Sorgu ölçümlerine genel bir bakış görüntülenir.
6. Sorgu profilini göster'e tıklayın.

Meta veri deponuzdan sorgu geçmişinin somutlaştırılması

Aşağıdaki kod, meta veri deposundan sorgu geçmişini oluşturmak için saatlik, günlük veya haftalık çalışan bir iş oluşturmak için kullanılabilir. HISTORY_TABLE_PATH ve LOOKUP_PERIOD_DAYS değişkenlerini buna göre ayarlayın.

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