Informations de référence sur la table système de travaux

Cet article fournit une référence sur l’utilisation des tables système lakeflow pour surveiller les travaux dans votre compte. Ces tables incluent des enregistrements de tous les espaces de travail de votre compte déployés dans la même région cloud. Pour afficher les enregistrements d’une autre région, vous devez afficher les tables d’un espace de travail déployé dans cette région.

system.lakeflow doivent être activées par un administrateur de compte. Vous pouvez les activer à l’aide de l’API SystemSchemas.

Pour obtenir des exemples d’utilisation de ces tables pour l’observabilité des coûts de travail et d’intégrité, consultez Surveiller les coûts de travail avec les tables système.

Tables de travail disponibles

Toutes les tables système liées aux travaux résident dans le system.lakeflow schéma. Actuellement, le schéma héberge quatre tables :

• jobs : effectue le suivi de la création, de la suppression et des informations de base des travaux.
• job_tasks : effectue le suivi de la création, de la suppression et des informations de base des tâches de travail.
• job_run_timeline : enregistre le début, la fin et l’état résultant des exécutions de travail.
• job_task_run_timeline : enregistre le début, la fin et l’état résultant des tâches de travail.

 : schéma de la table de travail.

jobs table est une table de dimension à modification lente. Lorsqu’une ligne change, une nouvelle ligne est émise, en remplaçant logiquement la ligne précédente.

Chemin d’accès de la table : cette table système se trouve à l’emplacement system.lakeflow.jobs.

Schéma de la table des tâches

La table des tâches de travail est une table de dimensions à variation lente. Lorsqu’une ligne change, une nouvelle ligne est émise, en remplaçant logiquement la ligne précédente.

Chemin d’accès de la table : cette table système se trouve à l’emplacement system.lakeflow.job_tasks.

Schéma de la table de chronologie d’exécution de travail

La table de chronologie de l’exécution du travail est immuable et terminée au moment où elle est produite.

Chemin d’accès de la table : cette table système se trouve à l’emplacement system.lakeflow.job_run_timeline.

Valeurs du type de déclencheur

Les valeurs possibles pour la trigger_type colonne sont les suivantes :

• CONTINUOUS
• CRON
• FILE_ARRIVAL
• ONETIME
• ONETIME_RETRY

Valeurs du type d’exécution

Les valeurs possibles pour la run_type colonne sont les suivantes :

• JOB_RUN
• SUBMIT_RUN : exécution unique créée via POST /api/2.1/jobs/runs/submit.
• WORKFLOW_RUN : exécution du travail lancée à partir du flux de travail du notebook.

Valeurs d’état de résultat

Les valeurs possibles pour la result_state colonne sont les suivantes :

• SUCCEEDED
• FAILED
• SKIPPED
• CANCELLED
• TIMED_OUT
• ERROR
• BLOCKED

Valeurs de code d’arrêt

Les valeurs possibles pour la termination_code colonne sont les suivantes :

• SUCCESS
• CANCELLED
• SKIPPED
• DRIVER_ERROR
• CLUSTER_ERROR
• REPOSITORY_CHECKOUT_FAILED
• INVALID_CLUSTER_REQUEST
• WORKSPACE_RUN_LIMIT_EXCEEDED
• FEATURE_DISABLED
• CLUSTER_REQUEST_LIMIT_EXCEEDED
• STORAGE_ACCESS_ERROR
• RUN_EXECUTION_ERROR
• UNAUTHORIZED_ERROR
• LIBRARY_INSTALLATION_ERROR
• MAX_CONCURRENT_RUNS_EXCEEDED
• MAX_SPARK_CONTEXTS_EXCEEDED
• RESOURCE_NOT_FOUND
• INVALID_RUN_CONFIGURATION
• CLOUD_FAILURE
• MAX_JOB_QUEUE_SIZE_EXCEEDED

Schéma de la table de chronologie exécutant la tâche de travail

La table de chronologie de la tâche de travail est immuable et terminée au moment où elle est produite.

Chemin d’accès de la table : cette table système se trouve à l’emplacement system.lakeflow.job_task_run_timeline.

Les valeurs possibles pour la result_state colonne sont les suivantes :

• SUCCEEDED
• FAILED
• SKIPPED
• CANCELLED
• TIMED_OUT
• ERROR
• BLOCKED

Les valeurs possibles pour la termination_code colonne sont les suivantes :

• SUCCESS
• CANCELLED
• SKIPPED
• DRIVER_ERROR
• CLUSTER_ERROR
• REPOSITORY_CHECKOUT_FAILED
• INVALID_CLUSTER_REQUEST
• WORKSPACE_RUN_LIMIT_EXCEEDED
• FEATURE_DISABLED
• CLUSTER_REQUEST_LIMIT_EXCEEDED
• STORAGE_ACCESS_ERROR
• RUN_EXECUTION_ERROR
• UNAUTHORIZED_ERROR
• LIBRARY_INSTALLATION_ERROR
• MAX_CONCURRENT_RUNS_EXCEEDED
• MAX_SPARK_CONTEXTS_EXCEEDED
• RESOURCE_NOT_FOUND
• INVALID_RUN_CONFIGURATION
• CLOUD_FAILURE
• MAX_JOB_QUEUE_SIZE_EXCEEDED

Exemples de requêtes

Cette section inclut des exemples de requêtes que vous pouvez utiliser pour tirer le meilleur parti des tables LakeFlow.

• Obtenir la version la plus récente des travaux
• Nombre quotidien de travaux par espace de travail
• Distribution quotidienne de l’état des travaux par espace de travail
• Vue d’ensemble des travaux les plus longs à exécuter
• Durée d’exécution du travail pour les travaux exécutés via runSubmit (c’est-à-dire Airflow)
• Analyse de l’exécution du travail
• Travaux en cours d’exécution sur le calcul à usage général
• Travail retenté s’exécute

Obtenir la version la plus récente des travaux

Les tables jobs et job_tasks modifient lentement les tables de dimension, par conséquent, un nouvel enregistrement est créé chaque fois qu’une modification est apportée. Pour obtenir la version la plus récente d’un travail, vous pouvez commander par colonne change_time.

SELECT
  *,
  ROW_NUMBER() OVER(PARTITION BY workspace_id, job_id ORDER BY change_time DESC) as rn
FROM
  system.lakeflow.jobs QUALIFY rn=1

Nombre de travaux quotidiens par espace de travail

Cette requête obtient le nombre quotidien de travaux par espace de travail pour les 7 derniers jours :

SELECT
  workspace_id,
  COUNT(DISTINCT run_id) as job_count,
  to_date(period_start_time) as date
FROM system.lakeflow.job_run_timeline
WHERE
  period_start_time > CURRENT_TIMESTAMP() - INTERVAL 7 DAYS
GROUP BY ALL

Distribution quotidienne de l’état des travaux par espace de travail

Cette requête retourne le nombre de travaux quotidiens par espace de travail pour les 7 derniers jours, distribués par le résultat de l’exécution du travail. La requête supprime tous les enregistrements dans lesquels les travaux sont dans un état en attente ou en cours d’exécution.

SELECT
  workspace_id,
  COUNT(DISTINCT run_id) as job_count,
  result_state,
  to_date(period_start_time) as date
FROM system.lakeflow.job_run_timeline
WHERE
  period_start_time > CURRENT_TIMESTAMP() - INTERVAL 7 DAYS
  AND result_state IS NOT NULL
GROUP BY ALL

Vue d’ensemble des travaux les plus longs à exécuter

Cette requête retourne le temps moyen d’exécutions du travail, mesuré en secondes. Les enregistrements sont organisés par travail. Une colonne de 90 et de 95 centiles supérieurs indique les longueurs moyennes des plus longues exécutions du travail.

with job_run_duration as (
    SELECT
        workspace_id,
        job_id,
        run_id,
        CAST(SUM(period_end_time - period_start_time) AS LONG) as duration
    FROM
        system.lakeflow.job_run_timeline
    WHERE
      period_start_time > CURRENT_TIMESTAMP() - INTERVAL 7 DAYS
    GROUP BY ALL
),
most_recent_jobs as (
  SELECT
    *,
    ROW_NUMBER() OVER(PARTITION BY workspace_id, job_id ORDER BY change_time DESC) as rn
  FROM
    system.lakeflow.jobs QUALIFY rn=1
)
SELECT
    t1.workspace_id,
    t1.job_id,
    first(t2.name, TRUE) as name,
    COUNT(DISTINCT t1.run_id) as runs,
    MEAN(t1.duration) as mean_seconds,
    AVG(t1.duration) as avg_seconds,
    PERCENTILE(t1.duration, 0.9) as p90_seconds,
    PERCENTILE(t1.duration, 0.95) as p95_seconds
FROM
    job_run_duration t1
    LEFT OUTER JOIN most_recent_jobs t2 USING (workspace_id, job_id)
GROUP BY ALL
ORDER BY mean_seconds DESC
LIMIT 100

Durée d’exécution du travail pour les travaux exécutés via runSubmit (c’est-à-dire Airflow)

Cette requête fournit une durée d’exécution historique pour un travail spécifique en fonction du paramètre run_name. Pour que la requête fonctionne, vous devez définir le run_name.

Vous pouvez également modifier la durée de l’analyse en mettant à jour le nombre de jours dans la section INTERVAL 60 DAYS.

SELECT
  workspace_id,
  run_id,
  SUM(period_end_time - period_start_time) as run_time
FROM system.lakeflow.job_run_timeline
WHERE
  run_type="SUBMIT_RUN"
  AND run_name={run_name}
  AND period_start_time > CURRENT_TIMESTAMP() - INTERVAL 60 DAYS
GROUP BY ALL

Analyse de l’exécution du travail

Cette requête fournit un runtime historique pour un travail spécifique. Pour que la requête fonctionne, vous devez définir un workspace_id et job_id.

Vous pouvez également modifier la durée de l’analyse en mettant à jour le nombre de jours dans la section INTERVAL 60 DAYS.

with job_run_duration as (
    SELECT
        workspace_id,
        job_id,
        run_id,
        min(period_start_time) as run_start,
        max(period_start_time) as run_end,
        CAST(SUM(period_end_time - period_start_time) AS LONG) as duration,
        FIRST(result_state, TRUE) as result_state
    FROM
        system.lakeflow.job_run_timeline
    WHERE
      period_start_time > CURRENT_TIMESTAMP() - INTERVAL 60 DAYS
      AND workspace_id={workspace_id}
      AND job_id={job_id}
    GROUP BY ALL
    ORDER BY run_start DESC
),
most_recent_jobs as (
  SELECT
    *,
    ROW_NUMBER() OVER(PARTITION BY workspace_id, job_id ORDER BY change_time DESC) as rn
  FROM
    system.lakeflow.jobs QUALIFY rn=1
)
SELECT
    t1.workspace_id,
    t1.job_id,
    t2.name,
    t1.run_id,
    t1.run_start,
    t1.run_end,
    t1.duration,
    t1.result_state
FROM job_run_duration t1
    LEFT OUTER JOIN most_recent_jobs t2 USING (workspace_id, job_id)

Travaux en cours d’exécution sur le calcul à usage général

Cette requête s’associe à la compute.clusters table système pour renvoyer les travaux récents exécutés sur le calcul à usage général au lieu du calcul des travaux.

with clusters AS (
  SELECT
    *,
    ROW_NUMBER() OVER(PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) as rn
  FROM system.compute.clusters
  WHERE cluster_source="UI" OR cluster_source="API"
  QUALIFY rn=1
),
job_tasks_exploded AS (
  SELECT
    workspace_id,
    job_id,
    EXPLODE(compute_ids) as cluster_id
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= CURRENT_DATE() - INTERVAL 30 DAY
),
all_purpose_cluster_jobs AS (
  SELECT
    t1.*,
    t2.cluster_name,
    t2.owned_by,
    t2.dbr_version
  FROM job_tasks_exploded t1
    INNER JOIN clusters t2 USING (workspace_id, cluster_id)
)
SELECT * FROM all_purpose_cluster_jobs LIMIT 10;

Travail retenté s’exécute

Cette requête collecte une liste d’exécutions de travaux retentées avec le nombre de nouvelles tentatives pour chaque exécution.

with repaired_runs as (
  SELECT
    workspace_id, job_id, run_id, COUNT(*) - 1 as retries_count
  FROM system.lakeflow.job_run_timeline
  WHERE result_state IS NOT NULL
  GROUP BY ALL
  HAVING retries_count > 0
)
SELECT
  *
FROM repaired_runs
ORDER BY retries_count DESC
LIMIT 10;