Consulte los conectores compatibles para la integración de aplicaciones.

Insertar datos en BigQuery mediante una tarea For Each Parallel

En este tutorial, creará una integración de aplicaciones y una subintegración para procesar una serie de registros. Para cada registro, la integración principal invoca asincrónicamente la subintegración, que toma los datos de cada registro y los inserta como una fila en una tabla de un conjunto de datos de BigQuery.

En este tutorial, completará las siguientes tareas:

Antes de empezar

  • Asegúrese de tener acceso a la integración de aplicaciones.

  • Haga lo siguiente en su proyecto de Google Cloud:

    • Otorgue los siguientes roles a la cuenta de servicio que desea utilizar para crear la conexión:
      • roles/bigquery.dataEditor
      • roles/bigquery.readSessionUser
      • roles/secretmanager.viewer
      • roles/secretmanager.secretAccessor
    • Habilitar los siguientes servicios:
      • secretmanager.googleapis.com (API del administrador de secretos)
      • connectors.googleapis.com (API de conectores)

      Si estos servicios no se han habilitado previamente para su proyecto, se le solicitará que los habilite al crear la conexión en la página Crear conexión.

Configurar una conexión de BigQuery

Empieza creando el conjunto de datos y la tabla de BigQuery que usarás en este tutorial. Una vez creados, crea una conexión de BigQuery. Usarás esta conexión en una integración más adelante en este tutorial.

Configurar un conjunto de datos y una tabla de BigQuery

Para configurar su conjunto de datos y tabla de BigQuery, realice los siguientes pasos:

  1. En la página de la consola de la nube , seleccione su Google Cloud proyecto.
  2. Para iniciar una sesión de Cloud Shell desde el Google Cloud consola, haga clic en elActivar el icono de Cloud ShellActiva el icono de Cloud Shell en la consola de Cloud . Esto inicia una sesión en el panel inferior de Google Cloud consola.
  3. Para habilitar las API de BigQuery, ingrese los siguientes comandos en su terminal de Cloud Shell:
    export PROJECT_ID=project_id
export REGION=region
gcloud services enable --project "${PROJECT_ID}" \
    bigquery.googleapis.com \
    bigquerystorage.googleapis.com
     En este comando, reemplace:
    • project_id con el ID del proyecto de su Google Cloud proyecto.
    • region con la región que desea utilizar para crear su conjunto de datos de BigQuery.
  4. Para crear un conjunto de datos de BigQuery con el nombre bq_tutorial , ingrese el siguiente comando en su terminal de Cloud Shell: 
          bq  --project_id ${PROJECT_ID} --location ${REGION} mk bq_tutorial
  5. Para crear una tabla de BigQuery con el nombre tutorial , ingresa el siguiente comando en tu terminal de Cloud Shell: 
          bq --project_id ${PROJECT_ID} \
        query  \
        --nouse_legacy_sql \
      'create table bq_tutorial.tutorial (
      unique_key STRING NOT NULL,
      created_date STRING,
      closed_date STRING,
      agency STRING,
      agency_name STRING,
      complaint_type STRING,
      descriptor STRING,
      location_type STRING,
      incident_zip STRING,
      incident_address STRING,
      street_name STRING,
      cross_street_1 STRING,
      cross_street_2 STRING,
      intersection_street_1 STRING,
      intersection_street_2 STRING,
      address_type STRING,
      city STRING,
      landmark STRING,
      facility_type STRING,
      status STRING,
      due_date STRING,
      resolution_action_updated_date STRING,
      community_board STRING,
      borough STRING,
      x_coordinate_state_plane STRING,
      y_coordinate_state_plane STRING,
      park_facility_name STRING,
      park_borough STRING,
      school_name STRING,
      school_number STRING,
      school_region STRING,
      school_code STRING,
      school_phone_number STRING,
      school_address STRING,
      school_city STRING,
      school_state STRING,
      school_zip STRING,
      school_not_found STRING,
      school_or_citywide_complaint STRING,
      vehicle_type STRING,
      taxi_company_borough STRING,
      taxi_pick_up_location STRING,
      bridge_highway_name STRING,
      bridge_highway_direction STRING,
      bridge_highway_segment STRING,
      road_ramp STRING,
      garage_lot_name STRING,
      ferry_direction STRING,
      ferry_terminal_name STRING,
      latitude STRING,
      longitude STRING,
      location STRING
      ) '
  6. Verify that your BigQuery table is created.
    1. In the Cloud console page, click the Navigation menu.
    2. In the Analytics section, click BigQuery.
    3. Expand your project and confirm that the bq_tutorial dataset is listed.
    4. Expand the bq_tutorial dataset and confirm that the tutorial table is listed.
    5. Click the documents table to view the schema.

Create a BigQuery connection

Next, you'll create a BigQuery connection. A BigQuery connection lets you insert, read, update and delete rows in a BigQuery table and use the resulting output in an integration. After creating the BigQuery connection, you'll use this connection in an integration later in this tutorial to add rows to the BigQuery table.

To create a BigQuery connection, complete the following steps:

  1. In the Cloud console page, select your Google Cloud project.
  2. Open the connections page.
  3. Click + CREATE NEW to open the Create Connection page.
  4. Configure the connection:
    1. In the Create Connection section, complete the following:
      • Connector: Select BigQuery from the drop down list of available Connectors.
      • Connector version: Select the latest Connector version from the drop down list of available versions.
      • In the Connection Name field, enter a name for the Connection instance. For this tutorial, enter connector-bq-tutorial.
      • Optionally, add a Description of the connection instance.
      • Service Account: Select a service account that has the required roles.
      • Project ID: Enter the ID of the Google Cloud project where the BigQuery data resides.
      • Dataset ID: Enter the ID of the BigQuery dataset that you want to use. For this tutorial, enter bq_tutorial.
      • Optionally, click + ADD LABEL to add a label in the form of a key/value pair.
      • Click Next.
    2. Location: Select a region from where the connection will run. Supported regions for connectors include:

        For the list of all the supported regions, see Locations.

      • Click Next.
    3. Authentication: The BigQuery connection does not require authentication configuration. Click Next.
    4. Review: Review your connection's configuration details. In this section, the connection and authentication details of the new connection are displayed for your review.
  5. Click Create.

Set up a sub-integration

In this tutorial, the sub-integration takes each record sent to it by the main integration and inserts it as a row in the tutorial table in the bq_tutorial dataset.

Create a sub-integration

To create the sub-integration, complete the following steps:

  1. In the Google Cloud console, go to the Application Integration page.

    Go to Application Integration

  2. Click Integrations from the left navigation menu to open the Integrations page.
  3. Click Create integration.
  4. In the Create Integration dialog, do the following:
    • Enter a name, for example, enter Process-each-record
    • Optionally, enter a description. For example, enter API Trigger to process each record (sub-integration)
    • Select the region where you want to create your integration.
  5. Click Create to open the integration editor.

Add an API Trigger

To add an API Trigger to the integration, do the following:

  1. In the integration editor, select Add a task/trigger > Triggers to display a list of available triggers.
  2. Drag the API Trigger element to the integration editor.

Add a Data Mapping task

To add a Data Mapping task in the integration, complete the following steps:

  1. Select +Add a task/trigger > Tasks in the integration editor to display the list of available tasks.
  2. Drag the Data Mapping element to the integration editor.

Configure the BigQuery connection

Now you are ready to use the BigQuery connection that you created earlier in the sub-integration. To configure the BigQuery connection in this integration, complete the following steps:

  1. Select +Add a task/trigger > Tasks in the integration editor to display the list of available tasks.
  2. Drag the Connectors element to the integration editor.
  3. Click the Connectors task element on the designer to view the task configuration pane.
  4. Click the edit icon on the right panel and update the Label to Insert row to BigQuery.
  5. Click Configure task.

    The Configure connector task dialog appears.

  6. In the Configure connector task dialog, do the following:
    1. Select the connection region where you created your BigQuery connection.
    2. Select the BigQuery connection that you want to use. For this tutorial, select connector-bq-tutorial.
    3. Once a connection is chosen, the Type column appears. Select Entities and then tutorial from the list of available entities.
    4. Once a type is chosen, the Operation column appears. Select Create.
    5. Click Done to complete the connection configuration and close the dialog.

Connect the integration elements

Next, add edge connections to connect the API Trigger to the Data Mapping task and the Data Mapping task to the Connectors task. An edge connection is a connection between any two elements in an integration. For more information on edges and edge conditions, see Edges.

To add the edge connections, complete the following steps:

  1. Click the Fork control point at the bottom of the API Trigger element. Drag and drop the edge connection at the Join control point at the top of the Data Mapping element.
  2. Click the Fork control point at the bottom of the Data Mapping element. Drag and drop the edge connection at the Join control point at the top of the Connectors element.

Configure the Data Mapping task

To configure the Data Mapping task, complete the following steps:

  1. In the integration editor, click the Data Mapping task to view the task configuration pane.
  2. Click Open Data Mapping Editor.
  3. In the Data Mapping Editor, click Add to add a new variable.
  4. In the Create Variable dialog, enter the following information:
    • Name: Enter record.
    • Data Type: Select JSON.
    • Schema: Select Infer from a sample JSON payload. Enter the following sample JSON payload: 
                  {
              "unique_key":"304271",
              "created_date":"02/06/2007 12:00:00 AM",
              "closed_date":"03/01/2007 12:00:00 AM",
              "agency":"TLC",
              "agency_name":"Taxi and Limousine Commission",
              "complaint_type":"Taxi Complaint",
              "descriptor":"Driver Complaint",
              "location_type":"Street",
              "incident_zip":"10001",
              "incident_address":"",
              "street_name":"",
              "cross_street_1":"",
              "cross_street_2":"",
              "intersection_street_1":"WEST 29 STREET",
              "intersection_street_2":"7 AVENUE",
              "address_type":"INTERSECTION",
              "city":"NEW YORK",
              "landmark":"",
              "facility_type":"N/A",
              "status":"Closed",
              "due_date":"02/28/2007 12:00:00 AM",
              "resolution_action_updated_date":"03/01/2007 12:00:00 AM",
              "community_board":"05 MANHATTAN",
              "borough":"MANHATTAN",
              "x_coordinate_state_plane":"986215",
              "y_coordinate_state_plane":"211740",
              "park_facility_name":"",
              "park_borough":"MANHATTAN",
              "school_name":"",
              "school_number":"",
              "school_region":"",
              "school_code":"",
              "school_phone_number":"",
              "school_address":"",
              "school_city":"",
              "school_state":"",
              "school_zip":"",
              "school_not_found":"",
              "school_or_citywide_complaint":"",
              "vehicle_type":"",
              "taxi_company_borough":"",
              "taxi_pick_up_location":"Other",
              "bridge_highway_name":"",
              "bridge_highway_direction":"",
              "road_ramp":"",
              "bridge_highway_segment":"",
              "garage_lot_name":"",
              "ferry_direction":"",
              "ferry_terminal_name":"",
              "latitude":"40.74785373937869",
              "longitude":"-73.99290823133913",
              "location":"(40.74785373937869, -73.99290823133913)"
            }
  5. Haga clic en Crear .
  6. Cuando se crea la variable, en el Editor de asignación de datos , complete los siguientes pasos:
    • Arrastre la nueva variable de registro a la columna Entrada .
    • Arrastre la variable conectorInputPayload a la columna Salida .
  7. Cierre el Editor de mapeo de datos para regresar al editor de integración.

Publicar la subintegración

Para publicar la subintegración, en el editor de integraciones, haga clic en Publicar .

Configurar la integración principal

En esta sección, se configura la integración principal, que utiliza una tarea "Para cada paralelo" para procesar cada registro. Esta integración principal invoca la subintegración una vez por cada registro.

Crear la integración principal

Para crear la integración principal, complete los siguientes pasos:

  1. En la consola de Google Cloud, vaya a la página Integración de aplicaciones .

    Ir a Integración de aplicaciones

  2. Haga clic en Integraciones en el menú de navegación de la izquierda para abrir la página Integraciones .
  3. Haga clic en Crear integración .
  4. En el cuadro de diálogo Crear integración , haga lo siguiente:
    • Introduzca un nombre, por ejemplo, introduzca registros-de-proceso .
    • Opcionalmente, introduzca una descripción. Por ejemplo, introduzca el disparador de API para procesar registros (integración principal).
    • Seleccione la región donde desea crear su integración.
  5. Haga clic en Crear para abrir el editor de integración.

Agregar un disparador de API

Para agregar un activador de API a la integración, haga lo siguiente:

  1. En el editor de integración, seleccione Agregar una tarea/desencadenador > Desencadenadores para mostrar una lista de desencadenadores disponibles.
  2. Arrastre el elemento API Trigger al editor de integración.

Agregar una tarea Para cada Paralelo

Para agregar una tarea For Each Parallel en la integración, complete los siguientes pasos:

  1. Seleccione +Agregar una tarea/desencadenante > Tareas en el editor de integración para mostrar la lista de tareas disponibles.
  2. Arrastre el elemento For Each Parallel al editor de integración.

Conectar los elementos de integración

A continuación, agregue una conexión de borde para conectar el disparador de API a la tarea For Each Parallel.

Para agregar la conexión de borde, haga clic en el punto de control "Bifurcación" en la parte inferior del elemento "Disparador de API". Arrastre y suelte la conexión de borde en el punto de control " Unión" en la parte superior del elemento de tarea "Para cada paralelo".

Configurar la tarea Para cada Paralelo

Para configurar la tarea Para cada paralelo, complete los siguientes pasos:

  1. En el editor de integración, haga clic en Para cada tarea paralela para ver el panel de configuración de tareas.
  2. En Selección de matriz > Lista para iterar , haga clic en Agregar nueva variable para agregar una nueva variable.
  3. En el cuadro de diálogo Crear variable , ingrese la siguiente información:
    • Nombre: Ingresar records
    • Tipo de datos: Seleccione JSON .
    • Esquema: Seleccione "Inferir desde una carga útil JSON de ejemplo" . Introduzca la siguiente carga útil JSON de ejemplo: 
                    [{
                "unique_key":"304271",
                "created_date":"02/06/2007 12:00:00 AM",
                "closed_date":"03/01/2007 12:00:00 AM",
                "agency":"TLC",
                "agency_name":"Taxi and Limousine Commission",
                "complaint_type":"Taxi Complaint",
                "descriptor":"Driver Complaint",
                "location_type":"Street",
                "incident_zip":"10001",
                "incident_address":"",
                "street_name":"",
                "cross_street_1":"",
                "cross_street_2":"",
                "intersection_street_1":"WEST 29 STREET",
                "intersection_street_2":"7 AVENUE",
                "address_type":"INTERSECTION",
                "city":"NEW YORK",
                "landmark":"",
                "facility_type":"N/A",
                "status":"Closed",
                "due_date":"02/28/2007 12:00:00 AM",
                "resolution_action_updated_date":"03/01/2007 12:00:00 AM",
                "community_board":"05 MANHATTAN",
                "borough":"MANHATTAN",
                "x_coordinate_state_plane":"986215",
                "y_coordinate_state_plane":"211740",
                "park_facility_name":"",
                "park_borough":"MANHATTAN",
                "school_name":"",
                "school_number":"",
                "school_region":"",
                "school_code":"",
                "school_phone_number":"",
                "school_address":"",
                "school_city":"",
                "school_state":"",
                "school_zip":"",
                "school_not_found":"",
                "school_or_citywide_complaint":"",
                "vehicle_type":"",
                "taxi_company_borough":"",
                "taxi_pick_up_location":"Other",
                "bridge_highway_name":"",
                "bridge_highway_direction":"",
                "road_ramp":"",
                "bridge_highway_segment":"",
                "garage_lot_name":"",
                "ferry_direction":"",
                "ferry_terminal_name":"",
                "latitude":"40.74785373937869",
                "longitude":"-73.99290823133913",
                "location":"(40.74785373937869, -73.99290823133913)"
              }]
  4. Haga clic en Crear .
  5. En la sección Detalles de subintegración , ingrese la siguiente información:
    • ID del disparador de API: Seleccione el elemento del disparador de API en la subintegración. Por ejemplo, seleccione Process-each-record_API_1 .
    • Estrategia de ejecución: Seleccionar ASYNC .
    • Seleccione Ejecutar una única integración .
  6. En la sección "En cada ejecución" , en "Dónde asignar elementos individuales de la matriz" , introduzca el nombre de la variable en la tarea de asignación de datos de la subintegración. En este caso, introduzca "registro" . Las variables de subintegración solo se muestran en las integraciones publicadas. Si no aparecen, actualice la página, ya que las variables tardan un tiempo en ser visibles después de publicar la subintegración.

Publicar la integración principal

Para publicar la integración principal, en el editor de integraciones, haga clic en Publicar .

Pon a prueba tu integración

Para probar su integración, complete los siguientes pasos:

  1. Descargue datos de muestra a su Cloud Shell:
    1. Para iniciar una sesión de Cloud Shell desde el Google Cloud consola, haga clic en elActivar el icono de Cloud ShellActiva el icono de Cloud Shell en la consola de Cloud . Esto inicia una sesión en el panel inferior de Google Cloud consola.
    2. Ingrese el siguiente comando en su terminal de Cloud Shell: 
      wget https://raw.githubusercontent.com/GoogleCloudPlatform/application-integration-samples/main/assets/bq-sample-dataset.json
    3. Para verificar que se descargaron los datos de muestra, ingrese el siguiente comando en su terminal de Cloud Shell:
      ls -la bq-sample-dataset.json
       El archivo descargado aparece en la terminal de Cloud Shell.
  2. Para seleccionar tres entradas aleatorias del conjunto de datos de muestra y almacenarlas de manera que pueda pasarlas a la integración, ingrese los siguientes comandos en su terminal de Cloud Shell: 
    AUTH=$(gcloud auth print-access-token)
export SAMPLE_DOCS=$(jq $(r=$((RANDOM % 1000)) ; echo ".[$r:$((r + 3))]") < bq-sample-dataset.json | jq -Rs '.')
            
generate_post_data()
  {
    cat <<EOF
      {
        "triggerId": "api_trigger/process-records_API_1",
        "inputParameters": 
          {
            "records": 
              {
                "jsonValue": $SAMPLE_DOCS
              }
          }
      }
      EOF
  }
  3. Para iniciar la prueba, ingrese el siguiente comando en su terminal de Cloud Shell:
    curl -X POST \
  https://integrations.googleapis.com/v1/projects/project_id/locations/region/integrations/process-records:execute \
  -H "Authorization: Bearer $AUTH" \
  -H "Content-Type: application/json" \
  -d "$(generate_post_data)"
     En este comando, reemplace:
    • project_id con el ID del proyecto de su Google Cloud proyecto.
    • region con la región donde creó su integración.
    Este comando invoca la integración principal y transfiere las entradas del conjunto de datos de muestra a la integración principal. Esta, a su vez, transfiere cada entrada a la subintegración, que agrega los datos como una fila en la tabla de BigQuery.
  4. Para verificar que su tabla de BigQuery ahora contiene estos registros, realice los siguientes pasos:
    1. En la página de la consola de Cloud , haga clic en el Menú de navegación .
    2. En la sección Análisis , haga clic en BigQuery .
    3. Expande tu proyecto y haz clic en el conjunto de datos bq_tutorial .
    4. Expanda el conjunto de datos bq_tutorial y haga clic en la tabla tutorial .
    5. Haga clic en la pestaña Explorador de tablas para ver los registros insertados.

¿Qué sigue?

Intenta crear integraciones con otros conectores. Para ver la lista de todos los conectores compatibles, consulta la referencia de conectores .