Créer des requêtes continues

Ce document explique comment exécuter une requête continue dans BigQuery.

Les requêtes continues BigQuery sont des instructions SQL qui s'exécutent en continu. Les requêtes continues vous permettent d'analyser les données entrantes dans BigQuery en temps réel, puis d'exporter les résultats vers Bigtable ou Pub/Sub, ou d'écrire les résultats dans une table BigQuery.

Sélectionner un type de compte

Vous pouvez créer et exécuter un job de requête continue à l'aide d'un compte utilisateur, ou créer un job de requête continue à l'aide d'un compte utilisateur, puis l'exécuter à l'aide d'un compte de service. Vous devez utiliser un compte de service pour exécuter une requête continue qui exporte les résultats vers un sujet Pub/Sub.

Lorsque vous utilisez un compte utilisateur, une requête continue s'exécute pendant deux jours maximum. Lorsque vous utilisez un compte de service, une requête continue s'exécute pendant 150 jours maximum. Pour en savoir plus, consultez la section Autorisation.

Autorisations requises

Cette section décrit les autorisations dont vous avez besoin pour créer et exécuter une requête continue. Au lieu des rôles IAM (Identity and Access Management) mentionnés, vous pouvez obtenir les autorisations requises via des rôles personnalisés.

Autorisations avec un compte utilisateur

Cette section fournit des informations sur les rôles et les autorisations nécessaires pour créer et exécuter une requête continue à l'aide d'un compte utilisateur.

Pour créer un job dans BigQuery, le compte utilisateur doit disposer de l'autorisation IAM bigquery.jobs.create. Chacun des rôles IAM suivants accorde l'autorisation bigquery.jobs.create :

• Utilisateur BigQuery (roles/bigquery.user)
• Utilisateur de job BigQuery (roles/bigquery.jobUser)
• Administrateur BigQuery (roles/bigquery.admin)

Pour exporter des données à partir d'une table BigQuery, le compte utilisateur doit disposer de l'autorisation IAM bigquery.tables.export. Chacun des rôles IAM suivants accorde l'autorisation bigquery.tables.export :

• Lecteur de données BigQuery (roles/bigquery.dataViewer)
• Éditeur de données BigQuery (roles/bigquery.dataEditor)
• Propriétaire de données BigQuery (roles/bigquery.dataOwner)
• Administrateur BigQuery (roles/bigquery.admin)

Pour mettre à jour les données d'une table BigQuery, le compte utilisateur doit disposer de l'autorisation IAM bigquery.tables.updateData. Chacun des rôles IAM suivants accorde l'autorisation bigquery.tables.updateData :

• Éditeur de données BigQuery (roles/bigquery.dataEditor)
• Propriétaire de données BigQuery (roles/bigquery.dataOwner)
• Administrateur BigQuery (roles/bigquery.admin)

Si le compte utilisateur doit activer les API nécessaires pour votre cas d'utilisation des requêtes continues, il doit disposer du rôle Administrateur Service Usage (roles/serviceusage.serviceUsageAdmin).

Autorisations avec un compte de service

Cette section fournit des informations sur les rôles et les autorisations nécessaires pour le compte utilisateur qui crée la requête continue, ainsi que pour le compte de service qui exécute la requête continue.

Autorisations du compte utilisateur

Pour créer un job dans BigQuery, le compte utilisateur doit disposer de l'autorisation IAM bigquery.jobs.create. Chacun des rôles IAM suivants accorde l'autorisation bigquery.jobs.create :

• Utilisateur BigQuery (roles/bigquery.user)
• Utilisateur de job BigQuery (roles/bigquery.jobUser)
• Administrateur BigQuery (roles/bigquery.admin)

Pour envoyer un job qui s'exécute à l'aide d'un compte de service, le compte utilisateur doit disposer du rôle Utilisateur du compte de service (roles/iam.serviceAccountUser). Si vous utilisez le même compte utilisateur pour créer le compte de service, celui-ci doit disposer du rôle Administrateur de compte de service (roles/iam.serviceAccountAdmin). Pour obtenir des informations sur la manière de limiter l'accès d'un utilisateur à un seul compte de service plutôt qu'à tous les comptes de service d'un projet, consultez la section Attribuer un rôle unique.

Si le compte utilisateur doit activer les API nécessaires pour votre cas d'utilisation des requêtes continues, il doit disposer du rôle Administrateur Service Usage (roles/serviceusage.serviceUsageAdmin).

Autorisations de compte de service

Pour exporter des données à partir d'une table BigQuery, le compte de service doit disposer de l'autorisation IAM bigquery.tables.export. Chacun des rôles IAM suivants accorde l'autorisation bigquery.tables.export :

• Lecteur de données BigQuery (roles/bigquery.dataViewer)
• Éditeur de données BigQuery (roles/bigquery.dataEditor)
• Propriétaire de données BigQuery (roles/bigquery.dataOwner)
• Administrateur BigQuery (roles/bigquery.admin)

• Éditeur de données BigQuery (roles/bigquery.dataEditor)
• Propriétaire de données BigQuery (roles/bigquery.dataOwner)
• Administrateur BigQuery (roles/bigquery.admin)

Avant de commencer

1. In the Trusted Cloud console, on the project selector page, select or create a Trusted Cloud project.

   Go to project selector

2. Make sure that billing is enabled for your Trusted Cloud project.

3. Enable the BigQuery API.

   Enable the API

Créer une réservation

Créez une réservation pour l'édition Enterprise ou Enterprise Plus, puis créez une attribution de réservation avec un type de tâche CONTINUOUS. Cette réservation peut utiliser l'autoscaling. Des limites de réservation s'appliquent aux attributions de réservation pour les requêtes continues.

Exporter vers Pub/Sub

Des API, des autorisations IAM et des ressources Trusted Cloud by S3NS supplémentaires sont nécessaires pour exporter des données vers Pub/Sub. Pour en savoir plus, consultez la section Exporter vers Pub/Sub.

Intégrer des attributs personnalisés sous forme de métadonnées dans des messages Pub/Sub

Vous pouvez utiliser des attributs Pub/Sub pour fournir des informations supplémentaires sur le message, telles que sa priorité, son origine, sa destination ou des métadonnées supplémentaires. Vous pouvez également utiliser des attributs pour filtrer les messages de l'abonnement.

Dans un résultat de requête continue, si une colonne est nommée _ATTRIBUTES, ses valeurs sont copiées dans les attributs de message Pub/Sub. Les champs fournis dans _ATTRIBUTES sont utilisés comme clés d'attribut.

La colonne _ATTRIBUTES doit être de type JSON, au format ARRAY<STRUCT<STRING, STRING>> ou STRUCT<STRING>.

Pour obtenir un exemple, consultez la section Exporter des données vers un sujet Pub/Sub.

Exporter vers Bigtable

Des API, des autorisations IAM et des ressources Trusted Cloudsupplémentaires sont nécessaires pour exporter des données vers Bigtable. Pour en savoir plus, consultez la section Exporter vers Bigtable.

Écrire des données dans une table BigQuery

Vous pouvez écrire des données dans une table BigQuery à l'aide d'une instruction INSERT.

Utiliser les fonctions d'IA

Des API, des autorisations IAM et des ressources Trusted Cloud supplémentaires sont nécessaires pour utiliser une fonction d'IA compatible dans une requête continue. Pour en savoir plus, consultez l'un des sujets suivants, en fonction de votre cas d'utilisation :

• Générer du texte à l'aide de la fonction ML.GENERATE_TEXT
• Générer des embeddings textuels à l'aide de la fonction ML.GENERATE_EMBEDDING
• Comprenez du texte avec la fonction ML.UNDERSTAND_TEXT
• Traduire du texte avec la fonction ML.TRANSLATE

Lorsque vous utilisez une fonction d'IA dans une requête continue, vérifiez si la sortie de la requête reste dans le quota de la fonction. Si vous dépassez le quota, vous devrez peut-être gérer séparément les enregistrements qui ne sont pas traités.

Spécifier un point de départ

Vous devez utiliser la fonction APPENDS dans la clause FROM d'une requête continue pour spécifier les données les plus anciennes à traiter. Par exemple, APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) indique à BigQuery de traiter les données ajoutées à la table my_table au plus 10 minutes avant le début de la requête continue. Les données ajoutées à my_table sont traitées au fur et à mesure. Aucun délai n'est imposé au traitement des données. Ne fournissez pas d'argument end_timestamp à la fonction APPENDS lorsque vous l'utilisez dans une requête continue.

L'exemple suivant montre comment lancer une requête continue à partir d'un moment précis à l'aide de la fonction APPENDS lorsque vous interrogez une table BigQuery qui reçoit des informations sur les trajets en taxi en streaming:

EXPORT DATA
  OPTIONS (format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides') AS (
  SELECT
    TO_JSON_STRING(STRUCT(ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE
    ride_status = 'enroute');

Spécifier un point de départ antérieur à l'heure actuelle

Si vous souhaitez traiter des données antérieures au moment actuel, vous pouvez utiliser la fonction APPENDS pour spécifier un point de départ antérieur pour la requête. Le point de départ que vous spécifiez doit se trouver dans la fenêtre de fonctionnalité temporelle de la table à partir de laquelle vous effectuez la sélection. La période temporelle couvre les sept derniers jours par défaut.

Pour inclure des données en dehors de la période de voyage dans le temps, utilisez une requête standard pour insérer ou exporter des données jusqu'à un moment précis, puis lancez une requête continue à partir de ce moment.

Exemple

L'exemple suivant montre comment charger des données plus anciennes à partir d'une table BigQuery qui reçoit des informations sur les trajets en taxi en streaming jusqu'à un moment donné dans une table, puis lancer une requête continue à partir du point d'arrêt des données plus anciennes.

1. Exécutez une requête standard pour remplir les données jusqu'à un moment donné:

   INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
   SELECT
     timestamp,
     meter_reading,
     ride_status,
     passenger_count,
     ST_Distance(
       ST_GeogPoint(pickup_longitude, pickup_latitude),
       ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
       SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
   FROM `myproject.real_time_taxi_streaming.taxirides`
     -- Include all data inserted into the table up to this point in time.
     -- This timestamp must be within the time travel window.
     FOR SYSTEM_TIME AS OF '2025-01-01 00:00:00 UTC'
   WHERE
     ride_status = 'dropoff';

2. Exécutez une requête continue à partir du moment où la requête s'est arrêtée:

   INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
   SELECT
     timestamp,
     meter_reading,
     ride_status,
     passenger_count,
     ST_Distance(
       ST_GeogPoint(pickup_longitude, pickup_latitude),
       ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
       SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
   FROM
     APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
       -- Configure the APPENDS TVF start_timestamp to start processing
       -- data right where the batch query left off.
       -- This timestamp must be within the time travel window.
       TIMESTAMP '2025-01-01 00:00:00 UTC' + INTERVAL 1 MICROSECOND)
   WHERE
     ride_status = 'dropoff';

Exécuter une requête continue à l'aide d'un compte utilisateur

Cette section explique comment exécuter une requête continue à l'aide d'un compte utilisateur. Une fois la requête continue exécutée, vous pouvez fermer la console, la fenêtre de terminal ou l'application sans interrompre l'exécution de la requête. Trusted Cloud Une requête continue exécutée par un compte utilisateur s'exécute pendant deux jours maximum, puis s'arrête automatiquement. Pour continuer à traiter les nouvelles données entrantes, lancez une nouvelle requête continue et spécifiez un point de départ. Pour automatiser ce processus, consultez la section Réessayer les requêtes ayant échoué.

Pour exécuter une requête continue, procédez comme suit :

curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true}}}' \
  --compressed

Exécuter une requête continue à l'aide d'un compte de service

Cette section explique comment exécuter une requête continue à l'aide d'un compte de service. Une fois la requête continue exécutée, vous pouvez fermer la console, la fenêtre de terminal ou l'application sans interrompre l'exécution de la requête. Trusted Cloud Une requête continue exécutée à l'aide d'un compte de service peut s'exécuter pendant 150 jours maximum, puis s'arrête automatiquement. Pour continuer à traiter les nouvelles données entrantes, lancez une nouvelle requête continue et spécifiez un point de départ. Pour automatiser ce processus, consultez la section Réessayer les requêtes ayant échoué.

Pour utiliser un compte de service afin d'exécuter une requête continue, procédez comme suit :

Créer un ID de tâche personnalisé

Chaque tâche de requête est associée à un ID de tâche que vous pouvez utiliser pour rechercher et gérer la tâche. Par défaut, les ID de tâche sont générés de manière aléatoire. Pour faciliter la recherche de l'ID de tâche d'une requête continue à l'aide de l'historique des tâches ou de l'explorateur de jobs, vous pouvez attribuer un préfixe d'ID de tâche personnalisé:

1. Dans la console Trusted Cloud , accédez à la page BigQuery.

   Accéder à BigQuery

2. Dans l'éditeur de requête, cliquez sur Plus.

3. Dans la section Choisir le mode de requête, sélectionnez Requête continue.

4. Cliquez sur Confirmer.

5. Dans l'éditeur de requête, cliquez sur Plus > Paramètres de requête.

6. Dans la section Préfixe d'ID de tâche personnalisé, saisissez un préfixe de nom personnalisé.

7. Cliquez sur Enregistrer.

Examples

Les exemples SQL suivants présentent des cas d'utilisation courants des requêtes continues.

Exporter des données vers un sujet Pub/Sub

L'exemple suivant montre une requête continue qui filtre les données d'une table BigQuery qui reçoit des flux d'informations sur les trajets en taxi, et publie les données sur un sujet Pub/Sub en temps réel avec des attributs de messages :

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message,
    TO_JSON(
      STRUCT(
        CAST(passenger_count AS STRING) AS passenger_count)) AS _ATTRIBUTES
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Exporter des données vers une table Bigtable

L'exemple suivant montre une requête continue qui filtre les données d'une table BigQuery qui reçoit des flux d'informations sur les trajets en taxi, et les exporte dans une table Bigtable en temps réel :

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_BIGTABLE',
    truncate = TRUE,
    overwrite = TRUE,
    uri = 'https://bigtable.googleapis.com/projects/myproject/instances/mybigtableinstance/tables/taxi-real-time-rides')
AS (
  SELECT
    CAST(CONCAT(ride_id, timestamp, latitude, longitude) AS STRING) AS rowkey,
    STRUCT(
      timestamp,
      latitude,
      longitude,
      meter_reading,
      ride_status,
      passenger_count) AS features
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Écrire des données dans une table BigQuery

L'exemple suivant montre une requête continue qui filtre et transforme les données d'une table BigQuery qui reçoit des flux d'informations sur les trajets en taxi, puis écrit les données dans une autre table BigQuery en temps réel. Les données sont ainsi disponibles pour une analyse plus approfondie en aval.

INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- Configure the APPENDS TVF start_timestamp to specify when you want to
    -- start processing data using your continuous query.
    -- This example starts processing at 10 minutes before the current time.
    CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
WHERE
  ride_status = 'dropoff';

Traiter des données à l'aide d'un modèle Vertex AI

L'exemple suivant montre une requête continue qui utilise un modèle Vertex AI pour générer une annonce pour les passagers de taxi en fonction de leur latitude et de leur longitude actuelles, puis qui exporte les résultats dans un sujet Pub/Sub en temps réel :

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude,
        prompt,
        ml_generate_text_llm_result)) AS message
  FROM
    ML.GENERATE_TEXT(
      MODEL `myproject.real_time_taxi_streaming.taxi_ml_generate_model`,
      (
        SELECT
          timestamp,
          ride_id,
          latitude,
          longitude,
          CONCAT(
            'Generate an ad based on the current latitude of ',
            latitude,
            ' and longitude of ',
            longitude) AS prompt
        FROM
          APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
            -- Configure the APPENDS TVF start_timestamp to specify when you
            -- want to start processing data using your continuous query.
            -- This example starts processing at 10 minutes before the current time.
            CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
        WHERE ride_status = 'enroute'
      ),
      STRUCT(
        50 AS max_output_tokens,
        1.0 AS temperature,
        40 AS top_k,
        1.0 AS top_p,
        TRUE AS flatten_json_output))
      AS ml_output
);

Modifier le code SQL d'une requête continue

Vous ne pouvez pas mettre à jour le code SQL utilisé dans une requête continue pendant l'exécution du job de requête continue. Vous devez annuler le job de requête continue, modifier le code SQL, puis démarrer un nouveau job de requête continue à partir du moment où vous avez arrêté le job de requête continue d'origine.

Pour modifier le code SQL utilisé dans une requête continue, procédez comme suit :

1. Affichez les détails du job pour le job de requête continue que vous souhaitez mettre à jour et notez l'ID du job.
2. Si possible, suspendez la collecte des données en amont. Si vous ne pouvez pas le faire, vous risquez de dupliquer des données lorsque la requête continue sera redémarrée.
3. Annulez la requête continue que vous souhaitez modifier.

4. Récupérez la valeur end_time du job de requête continue d'origine à l'aide de la vue JOBS INFORMATION_SCHEMA :

   SELECT end_time
   FROM `PROJECT_ID.region-REGION`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
   WHERE
     EXTRACT(DATE FROM creation_time) = current_date()
   AND error_result.reason = 'stopped'
   AND job_id = 'JOB_ID';

   Remplacez les éléments suivants :

   • PROJECT_ID : ID de votre projet.
   • REGION : région utilisée par votre projet.
   • JOB_ID : ID du job de requête continue que vous avez identifié à l'étape 1.

5. Modifiez l'instruction SQL de requête continue pour démarrer la requête continue à partir d'un moment précis, en utilisant la valeur end_time récupérée à l'étape 5 comme point de départ.

6. Modifiez l'instruction SQL de la requête continue pour refléter les modifications nécessaires.

7. Exécutez la requête continue modifiée.

Annuler une requête continue

Vous pouvez annuler un job de requête continue comme n'importe quel autre job. L'arrêt de l'exécution du job peut prendre jusqu'à une minute après l'annulation du job.

Si vous annulez une requête, puis la redémarrez, elle se comporte comme une nouvelle requête indépendante. La requête redémarrée ne commence pas à traiter les données à l'endroit où la tâche précédente s'est arrêtée et ne peut pas faire référence aux résultats de la requête précédente. Consultez la section Démarrer une requête continue à partir d'un moment précis.

Surveiller les requêtes et gérer les erreurs

Une requête continue peut être interrompue en raison de facteurs tels que des incohérences de données, des modifications de schéma, des perturbations temporaires du service ou une maintenance. Bien que BigQuery gère certaines erreurs temporaires, voici quelques bonnes pratiques pour améliorer la résilience des tâches:

• Surveillez les requêtes continues.
• Alerte en cas de requêtes ayant échoué
• Réessayer les requêtes ayant échoué.

Étapes suivantes

• Surveiller les requêtes continues