Crear consultas continuas

En este documento se describe cómo ejecutar una consulta continua en BigQuery.

Las consultas continuas de BigQuery son instrucciones SQL que se ejecutan de forma continua. Las consultas continuas te permiten analizar los datos entrantes en BigQuery en tiempo real y, a continuación, exportar los resultados a Bigtable, Pub/Sub o Spanner, o bien escribir los resultados en una tabla de BigQuery.

Elige un tipo de cuenta

Puedes crear y ejecutar un trabajo de consulta continua con una cuenta de usuario, o bien crear un trabajo de consulta continua con una cuenta de usuario y, a continuación, ejecutarlo con una cuenta de servicio. Debes usar una cuenta de servicio para ejecutar una consulta continua que exporte resultados a un tema de Pub/Sub.

Si usas una cuenta de usuario, la consulta continua se ejecuta durante un máximo de dos días. Si usas una cuenta de servicio, una consulta continua se ejecuta durante un máximo de 150 días. Para obtener más información, consulta la sección Autorización.

Permisos obligatorios

En esta sección se describen los permisos que necesitas para crear y ejecutar una consulta continua. Como alternativa a los roles de Gestión de Identidades y Accesos (IAM) mencionados, puedes obtener los permisos necesarios a través de roles personalizados.

Permisos al usar una cuenta de usuario

En esta sección se proporciona información sobre los roles y permisos necesarios para crear y ejecutar una consulta continua con una cuenta de usuario.

Para crear un trabajo en BigQuery, la cuenta de usuario debe tener el permiso de gestión de identidades y accesos bigquery.jobs.create. Cada uno de los siguientes roles de gestión de identidades y accesos concede el permiso bigquery.jobs.create:

• Usuario de BigQuery (roles/bigquery.user)
• Usuario de trabajo de BigQuery (roles/bigquery.jobUser)
• Administrador de BigQuery (roles/bigquery.admin)

Para exportar datos de una tabla de BigQuery, la cuenta de usuario debe tener el permiso de gestión de identidades y accesos bigquery.tables.export . Cada uno de los siguientes roles de gestión de identidades y accesos concede el permiso bigquery.tables.export:

• Lector de datos de BigQuery (roles/bigquery.dataViewer)
• Editor de datos de BigQuery (roles/bigquery.dataEditor)
• Propietario de datos de BigQuery (roles/bigquery.dataOwner)
• Administrador de BigQuery (roles/bigquery.admin)

Para actualizar los datos de una tabla de BigQuery, la cuenta de usuario debe tener el permiso de gestión de identidades y accesos bigquery.tables.updateData. Cada uno de los siguientes roles de gestión de identidades y accesos concede el permiso bigquery.tables.updateData:

• Editor de datos de BigQuery (roles/bigquery.dataEditor)
• Propietario de datos de BigQuery (roles/bigquery.dataOwner)
• Administrador de BigQuery (roles/bigquery.admin)

Si la cuenta de usuario debe habilitar las APIs necesarias para tu caso práctico de consulta continua, debe tener el rol Administrador de uso de servicios (roles/serviceusage.serviceUsageAdmin).

Permisos al usar una cuenta de servicio

En esta sección se proporciona información sobre los roles y permisos que necesitan la cuenta de usuario que crea la consulta continua y la cuenta de servicio que ejecuta la consulta continua.

Permisos de cuenta de usuario

Para crear un trabajo en BigQuery, la cuenta de usuario debe tener el permiso de gestión de identidades y accesos bigquery.jobs.create. Cada uno de los siguientes roles de gestión de identidades y accesos concede el permiso bigquery.jobs.create:

• Usuario de BigQuery (roles/bigquery.user)
• Usuario de trabajo de BigQuery (roles/bigquery.jobUser)
• Administrador de BigQuery (roles/bigquery.admin)

Para enviar un trabajo que se ejecute con una cuenta de servicio, la cuenta de usuario debe tener el rol Usuario de cuenta de servicio (roles/iam.serviceAccountUser). Si usas la misma cuenta de usuario para crear la cuenta de servicio, esta debe tener el rol Administrador de cuentas de servicio (roles/iam.serviceAccountAdmin). Para obtener información sobre cómo limitar el acceso de un usuario a una sola cuenta de servicio en lugar de a todas las cuentas de servicio de un proyecto, consulta Asignar un solo rol.

Si la cuenta de usuario debe habilitar las APIs necesarias para tu caso práctico de consulta continua, debe tener el rol Administrador de uso de servicios (roles/serviceusage.serviceUsageAdmin).

Permisos de cuenta de servicio

Para exportar datos de una tabla de BigQuery, la cuenta de servicio debe tener el permiso de gestión de identidades y accesos bigquery.tables.export. Cada uno de los siguientes roles de gestión de identidades y accesos concede el permiso bigquery.tables.export:

• Lector de datos de BigQuery (roles/bigquery.dataViewer)
• Editor de datos de BigQuery (roles/bigquery.dataEditor)
• Propietario de datos de BigQuery (roles/bigquery.dataOwner)
• Administrador de BigQuery (roles/bigquery.admin)

• Editor de datos de BigQuery (roles/bigquery.dataEditor)
• Propietario de datos de BigQuery (roles/bigquery.dataOwner)
• Administrador de BigQuery (roles/bigquery.admin)

Antes de empezar

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

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Go to project selector

2. Verify that billing is enabled for your Google Cloud project.

3. Enable the BigQuery API.

   Roles required to enable APIs

   To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

   Enable the API

Crear una reserva

Crea una reserva de la edición Enterprise o Enterprise Plus y, a continuación, crea una asignación de reserva con el tipo de trabajo CONTINUOUS. Esta reserva puede usar el escalado automático. Hay limitaciones de reserva que se aplican a las asignaciones de reservas de consultas continuas.

Exportar a Pub/Sub

Se necesitan APIs, permisos de IAM y Google Cloud recursos adicionales para exportar datos a Pub/Sub. Para obtener más información, consulta el artículo Exportar a Pub/Sub.

Insertar atributos personalizados como metadatos en mensajes de Pub/Sub

Puede usar atributos de Pub/Sub para proporcionar información adicional sobre el mensaje, como su prioridad, origen, destino o metadatos adicionales. También puedes usar atributos para filtrar mensajes de la suscripción.

En el resultado de una consulta continua, si una columna se llama _ATTRIBUTES, sus valores se copian en los atributos del mensaje de Pub/Sub. Los campos proporcionados en _ATTRIBUTES se usan como claves de atributo.

La columna _ATTRIBUTES debe ser de tipo JSON y tener el formato ARRAY<STRUCT<STRING, STRING>> o STRUCT<STRING>.

Para ver un ejemplo, consulta Exportar datos a un tema de Pub/Sub.

Exportar a Bigtable

Se necesitan APIs, permisos de IAM y Google Cloud recursos adicionales para exportar datos a Bigtable. Para obtener más información, consulta Exportar a Bigtable.

Exportar a Spanner

Se necesitan APIs, permisos de IAM y recursos Google Cloudadicionales para exportar datos a Spanner. Para obtener más información, consulta el artículo Exportar a Spanner (ETL inverso).

Escribir datos en una tabla de BigQuery

Puedes escribir datos en una tabla de BigQuery mediante una instrucción INSERT.

Usar funciones de IA

Para usar una función de IA compatible en una consulta continua, se necesitan APIs, permisos de gestión de identidades y accesos y Google Cloud recursos adicionales. Para obtener más información, consulta uno de los siguientes temas, en función de tu caso práctico:

• Generar texto con la función ML.GENERATE_TEXT
• Generar incrustaciones de texto con la función ML.GENERATE_EMBEDDING
• Interpretar texto con la función ML.UNDERSTAND_TEXT
• Traducir texto con la función ML.TRANSLATE

Cuando uses una función de IA en una consulta continua, ten en cuenta si el resultado de la consulta se mantendrá dentro de la cuota de la función. Si supera la cuota, es posible que tenga que gestionar por separado los registros que no se procesen.

Especificar un punto de partida

Debes usar la función APPENDS en la cláusula FROM de una consulta continua para especificar los datos más antiguos que se van a procesar. Por ejemplo, APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) indica a BigQuery que procese los datos que se añadieron a la tabla my_table como máximo 10 minutos antes del inicio de la consulta continua. Los datos que se añaden a my_table se procesan a medida que se reciben. No se impone ningún retraso en el tratamiento de los datos. No proporciones un argumento end_timestamp a la función APPENDS cuando la uses en una consulta continua.

En el siguiente ejemplo se muestra cómo iniciar una consulta continua desde un momento concreto mediante la función APPENDS al consultar una tabla de BigQuery que recibe información de viajes 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');

Especificar un punto de partida anterior a la hora actual

Si quiere procesar datos de un momento anterior al actual, puede usar la función APPENDS para especificar un punto de partida anterior para la consulta. El punto de partida que especifiques debe estar dentro del periodo de viaje en el tiempo de la tabla que selecciones. De forma predeterminada, la ventana de viaje en el tiempo abarca los últimos siete días.

Para incluir datos que estén fuera del periodo de la función de viaje en el tiempo, usa una consulta estándar para insertar o exportar datos hasta un momento concreto y, a continuación, inicia una consulta continua desde ese momento.

Ejemplo

En el siguiente ejemplo se muestra cómo cargar datos antiguos de una tabla de BigQuery que recibe información de viajes en taxi en streaming hasta un momento concreto en una tabla y, a continuación, iniciar una consulta continua desde el punto de corte de los datos antiguos.

1. Ejecuta una consulta estándar para rellenar los datos hasta un momento concreto:

   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. Ejecuta una consulta continua desde el momento en el que se detuvo:

   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';

Ejecutar una consulta continua con una cuenta de usuario

En esta sección se describe cómo ejecutar una consulta continua mediante una cuenta de usuario. Una vez que se esté ejecutando la consulta continua, puedes cerrar la Google Cloud consola, la ventana de terminal o la aplicación sin interrumpir la ejecución de la consulta. Una consulta continua ejecutada por una cuenta de usuario se ejecuta durante un máximo de dos días y, después, se detiene automáticamente. Para seguir procesando los nuevos datos entrantes, inicia una nueva consulta continua y especifica un punto de partida. Para automatizar este proceso, consulta Reintentar consultas fallidas.

Sigue estos pasos para ejecutar una consulta continua:

Ejecutar una consulta continua con una cuenta de servicio

En esta sección se describe cómo ejecutar una consulta continua mediante una cuenta de servicio. Una vez que se esté ejecutando la consulta continua, puedes cerrar la Google Cloud consola, la ventana de terminal o la aplicación sin interrumpir la ejecución de la consulta. Una consulta continua que se ejecuta mediante una cuenta de servicio puede durar hasta 150 días y, después, se detiene automáticamente. Para seguir procesando los nuevos datos entrantes, inicia una nueva consulta continua y especifica un punto de partida. Para automatizar este proceso, consulta Reintentar consultas fallidas.

Sigue estos pasos para usar una cuenta de servicio y ejecutar una consulta continua:

Crear un ID de tarea personalizado

A cada trabajo de consulta se le asigna un ID que puede usar para buscarlo y gestionarlo. De forma predeterminada, los IDs de trabajo se generan aleatoriamente. Para que te resulte más fácil buscar el ID de una consulta continua mediante el historial de tareas o el explorador de tareas, puedes asignar un prefijo de ID de tarea personalizado:

1. En la Google Cloud consola, ve a la página BigQuery.

   Ir a BigQuery

2. En el editor de consultas, haz clic en Más.

3. En la sección Elegir modo de consulta, selecciona Consulta continua.

4. Haz clic en Confirmar.

5. En el editor de consultas, haz clic en Más > Configuración de consultas.

6. En la sección Prefijo de ID de trabajo personalizado, introduce un prefijo de nombre personalizado.

7. Haz clic en Guardar.

Ejemplos

En los siguientes ejemplos de SQL se muestran casos prácticos habituales de consultas continuas.

Exportar datos a un tema de Pub/Sub

En el siguiente ejemplo se muestra una consulta continua que filtra datos de una tabla de BigQuery que recibe información de viajes en taxi en streaming y publica los datos en un tema de Pub/Sub en tiempo real con atributos de mensaje:

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'
);

Exportar datos a una tabla de Bigtable

En el siguiente ejemplo se muestra una consulta continua que filtra datos de una tabla de BigQuery que recibe información de viajes en taxi en streaming y exporta los datos a una tabla de Bigtable en tiempo real:

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'
);

Exportar datos a una tabla de Spanner

En el siguiente ejemplo se muestra una consulta continua que filtra datos de una tabla de BigQuery que recibe información de viajes en taxi en streaming y, a continuación, exporta los datos a una tabla de Spanner en tiempo real:

EXPORT DATA
 OPTIONS (
   format = 'CLOUD_SPANNER',
   uri = 'https://spanner.googleapis.com/projects/myproject/instances/myspannerinstance/databases/taxi-real-time-rides',
   spanner_options ="""{
      "table": "rides",
      -- To ensure data is written to Spanner in the correct sequence
      -- during a continuous export, use the change_timestamp_column
      -- option. This should be mapped to a timestamp column from your
      -- BigQuery data. If your source data lacks a timestamp, the 
      -- _CHANGE_TIMESTAMP pseudocolumn provided by the APPENDS function 
      -- will be automatically mapped to the "change_timestamp" column.
      "change_timestamp_column": "change_timestamp"
   }"""
  )
  AS (
  SELECT
    ride_id,
    latitude,
    longitude,
    meter_reading,
    ride_status,
    passenger_count
  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'
  );

Escribir datos en una tabla de BigQuery

En el siguiente ejemplo se muestra una consulta continua que filtra y transforma datos de una tabla de BigQuery que recibe información de viajes en taxi en streaming y, a continuación, escribe los datos en otra tabla de BigQuery en tiempo real. De esta forma, los datos estarán disponibles para realizar análisis posteriores.

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';

Procesar datos con un modelo de Vertex AI

En el siguiente ejemplo se muestra una consulta continua que usa un modelo de Vertex AI para generar un anuncio para los pasajeros de taxi en función de su latitud y longitud actuales. A continuación, exporta los resultados a un tema de Pub/Sub en tiempo real:

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

Modificar el SQL de una consulta continua

No puedes actualizar el SQL usado en una consulta continua mientras se esté ejecutando el trabajo de la consulta continua. Debes cancelar el trabajo de consulta continua, modificar el SQL y, a continuación, iniciar un nuevo trabajo de consulta continua desde el punto en el que detuviste el trabajo de consulta continua original.

Para modificar el SQL usado en una consulta continua, sigue estos pasos:

1. Consulta los detalles del trabajo de la consulta continua que quieras actualizar y anota el ID del trabajo.
2. Si es posible, pausa la recogida de datos de nivel superior. Si no puedes hacerlo, es posible que se dupliquen algunos datos cuando se reinicie la consulta continua.
3. Cancela la consulta continua que quieras modificar.

4. Obtén el valor end_time del trabajo de consulta continua original mediante la vista 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';

   Haz los cambios siguientes:

   • PROJECT_ID: tu ID de proyecto.
   • REGION: la región que usa tu proyecto.
   • JOB_ID: el ID del trabajo de consulta continua que has identificado en el paso 1.

5. Modifica la instrucción SQL de la consulta continua para iniciar la consulta continua desde un momento concreto. Para ello, usa el valor end_time que has obtenido en el paso 5 como punto de inicio.

6. Modifica la instrucción SQL de la consulta continua para reflejar los cambios que necesites.

7. Ejecuta la consulta continua modificada.

Cancelar una consulta continua

Puedes cancelar una consulta continua de la misma forma que cualquier otro trabajo. La consulta puede tardar hasta un minuto en dejar de ejecutarse después de que se cancele la tarea.

Si cancelas una consulta y la reinicias, la consulta reiniciada se comportará como una consulta nueva e independiente. La consulta reiniciada no empieza a procesar los datos en el punto en el que se detuvo la tarea anterior y no puede hacer referencia a los resultados de la consulta anterior. Consulta cómo iniciar una consulta continua desde un momento concreto.

Monitorizar consultas y gestionar errores

Una consulta continua puede interrumpirse debido a factores como incoherencias en los datos, cambios en el esquema, interrupciones temporales del servicio o tareas de mantenimiento. Aunque BigQuery gestiona algunos errores transitorios, las prácticas recomendadas para mejorar la resiliencia de las tareas incluyen lo siguiente:

• Monitorizar consultas continuas.
• Alertas sobre consultas fallidas.
• Reintenta las consultas fallidas.

Siguientes pasos

• Monitorizar consultas continuas