Inserisci i dati in BigQuery utilizzando un'attività per ogni parallelo

In questo tutorial creerai Apigee Integration e una sottointegrazione per elaborare una serie di record. Per ogni record, l'integrazione principale richiama in modo asincrono la sottointegrazione, che prende i dati per ogni record e li inserisce come riga in una tabella in un set di dati BigQuery.

In questo tutorial completerai le seguenti attività:

Prima di iniziare

  • Assicurati di avere accesso ad Apigee Integration.

  • Effettua le seguenti operazioni nel progetto Google Cloud:

    • Concedi i ruoli seguenti all'account di servizio che vuoi utilizzare per creare la connessione:
      • roles/bigquery.dataEditor
      • roles/bigquery.readSessionUser
      • roles/secretmanager.viewer
      • roles/secretmanager.secretAccessor
    • Attiva i seguenti servizi:
      • secretmanager.googleapis.com (API Secret Manager)
      • connectors.googleapis.com (API Connectors)

      Se questi servizi non sono stati abilitati per il tuo progetto in precedenza, ti verrà chiesto di abilitarli quando crei la connessione nella pagina Crea connessione.

Configura una connessione BigQuery

Per iniziare, crea il set di dati e la tabella BigQuery da utilizzare in questo tutorial. Dopo aver creato il set di dati e la tabella, crea una connessione BigQuery. Userai questa connessione in un'integrazione più avanti in questo tutorial.

configura un set di dati e una tabella BigQuery

Per configurare il set di dati e la tabella BigQuery, segui questi passaggi:

  1. Nella pagina della console Cloud, seleziona il tuo progetto Google Cloud.
  2. Per avviare una sessione di Cloud Shell dalla console Google Cloud, fai clic sull'icona Icona Attiva Cloud Shell Attiva Cloud Shell nella console Cloud. Viene avviata una nel riquadro inferiore della console Google Cloud.
  3. Per abilitare le API BigQuery, inserisci i comandi seguenti nel terminale Cloud Shell: 
    export PROJECT_ID=project_id
export REGION=region
gcloud services enable --project "${PROJECT_ID}" \
    bigquery.googleapis.com \
    bigquerystorage.googleapis.com
     In questo comando, sostituisci:
    • project_id con l'ID del tuo progetto Google Cloud.
    • region con la regione che vuoi utilizzare per creare il tuo set di dati BigQuery.
  4. Per creare un set di dati BigQuery con il nome bq_tutorial, inserisci il seguente comando nel terminale Cloud Shell: 
          bq  --project_id ${PROJECT_ID} --location ${REGION} mk bq_tutorial
  5. Per creare una tabella BigQuery con il nome tutorial, inserisci il comando seguente nel terminale 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 Apigee UI, select your Apigee Organization.
  2. Click Develop > Integrations.
  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)"
            }
    • Fai clic su Crea.
    • Dopo aver creato la variabile, completa i seguenti passaggi nell'editor mappatura dati:
      • Trascina la nuova variabile record nella colonna Input.
      • Trascina la variabile connectorInputPayload nella colonna Output.
    • Chiudi l'editor di mappatura dei dati per tornare all'editor dell'integrazione.

Pubblica la sottointegrazione

Per pubblicare l'integrazione secondaria, fai clic su Pubblica nell'editor delle integrazioni.

Configura l'integrazione principale

In questa sezione configurerai l'integrazione principale, che utilizza un'attività Per ogni parallelo per elaborare ogni record. Questa integrazione principale richiama quindi la sottointegrazione una volta per ogni record.

Crea l'integrazione principale

Per creare l'integrazione principale, completa i seguenti passaggi:

  1. Nell'interfaccia utente di Apigee, seleziona la tua organizzazione Apigee.
  2. Fai clic su Sviluppo > Integrazioni.
  3. Fai clic su Crea integrazione.
  4. Nella finestra di dialogo Crea integrazione, procedi nel seguente modo:
    • Inserisci un nome, ad esempio process-records.
    • Se vuoi, inserisci una descrizione. Ad esempio, inserisci Trigger API per l'elaborazione dei record (integrazione principale)
    • Seleziona la regione in cui vuoi creare l'integrazione.
  5. Fai clic su Crea per aprire l'editor delle integrazioni.

Aggiungi un trigger API

Per aggiungere un trigger API all'integrazione, segui questi passaggi:

  1. Nell'editor di integrazione, seleziona Aggiungi un'attività/un trigger > Trigger per visualizzare un elenco degli attivatori disponibili.
  2. Trascina l'elemento API Trigger nell'editor dell'integrazione.

Aggiungi un'attività per ogni attività parallela

Per aggiungere un'attività per ogni attività parallela nell'integrazione, completa i seguenti passaggi:

  1. Seleziona + Aggiungi un'attività/un attivatore > Tasks nell'editor dell'integrazione per visualizzare l'elenco delle attività disponibili.
  2. Trascina l'elemento Per ogni parallelo nell'editor dell'integrazione.

Collega gli elementi di integrazione

Poi, aggiungi una connessione perimetrale per connettere il trigger API all'attività Per ogni attività parallela.

Per aggiungere la connessione perimetrale, fai clic sul punto di controllo Fork nella parte inferiore dell'elemento trigger API. Trascina la connessione perimetrale in corrispondenza del punto di controllo Unisci nella parte superiore dell'elemento dell'attività Per ogni attività parallela.

Configura l'attività Per ogni attività parallela

Per configurare l'attività Per ogni attività parallela, completa i seguenti passaggi:

  1. Nell'editor dell'integrazione, fai clic sull'attività Per ogni parallelo per visualizzare il riquadro di configurazione dell'attività.
  2. In Selezione array > Elenco da ripetere: fai clic su Aggiungi nuova variabile per aggiungere una nuova variabile.
  3. Nella finestra di dialogo Crea variabile, inserisci le seguenti informazioni:
    • Nome: inserisci records
    • Tipo di dati:seleziona JSON.
    • Schema: seleziona Deduci da un payload JSON di esempio. Inserisci il seguente payload JSON di esempio: 
                    [{
                "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. Fai clic su Crea.
  5. Nella sezione Dettagli della sottointegrazione, inserisci le seguenti informazioni:
    • ID trigger API:seleziona l'elemento Trigger API nell'integrazione secondaria. Ad esempio, seleziona Process-each-record_API_1.
    • Strategia di esecuzione: seleziona ASYNC.
    • Seleziona Esegui una singola integrazione.
  6. Nella sezione A ogni esecuzione, in Dove mappare i singoli elementi dell'array, inserisci il nome della variabile nell'attività di mappatura dei dati nella sottointegrazione. In questo caso, inserisci record. Le variabili di integrazione secondaria sono elencate solo per le integrazioni pubblicate. Se le variabili non sono elencate, aggiorna la pagina, poiché è necessario del tempo prima che le variabili diventino visibili dopo la pubblicazione della sottointegrazione.

Pubblica l'integrazione principale

Per pubblicare l'integrazione principale, fai clic su Pubblica nell'editor delle integrazioni.

Crea un proxy API Apigee

Per attivare l'integrazione, crea un proxy API Apigee con la tua integrazione come destinazione. A tale scopo, completa i seguenti passaggi:

  1. Crea un account di servizio nel tuo progetto Google Cloud e assegnagli il ruolo richiesto, Apigee Integration Invoker. Per informazioni sull'assegnazione dei ruoli IAM, consulta Ruoli e autorizzazioni IAM.
  2. Accedi all'UI di Apigee.
  3. Nella barra di navigazione, seleziona Sviluppo > Proxy API.
  4. Fai clic su Crea nuovo.
  5. Nella procedura guidata Crea proxy, fai clic su Target di integrazione.
  6. Nella pagina Dettagli del proxy, inserisci le informazioni seguenti:
    • Nome: inserisci process-records.
    • Base path: inserisci /v1/process-records.
    • Regione di integrazione:seleziona la regione utilizzata per creare l'integrazione.
    • Target dell'integrazione: seleziona l'integrazione principale che hai creato. Per questo tutorial, seleziona process-records.
    • Trigger: seleziona il trigger API richiesto. Per questo tutorial, seleziona process-records_API_1.
    • Tipo di endpoint: seleziona Sincronizzazione.
  7. Fai clic su Avanti e fai clic di nuovo su Avanti nella pagina Criteri comuni.
  8. Nella pagina Riepilogo, fai clic su Crea.
  9. Quando il proxy viene creato, fai clic su Modifica proxy.
  10. Fai clic sulla scheda Sviluppo.
  11. In Criteri, aggiorna il criterio Set Integration Request (Imposta richiesta di integrazione) con il criterio seguente: 
    <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<SetIntegrationRequest continueOnError="false" enabled="true" name="set-integration-request">
  <DisplayName>Set Integration Request</DisplayName>
  <ProjectId>project_id</ProjectId>
  <IntegrationName>process-records</IntegrationName>
  <IntegrationRegion>region</IntegrationRegion>
  <ApiTrigger>api_trigger/process-records_API_1</ApiTrigger>
  <Parameters>
    <Parameter name="records" type="json" ref="request.content"/>
  </Parameters>
</SetIntegrationRequest>
     Sostituisci:
    • project_id con l'ID del tuo progetto Google Cloud.
    • region con la regione in cui hai creato l'integrazione.
  12. Salva le modifiche.
  13. Fai clic su Esegui il deployment.
  14. In Revisione, seleziona la revisione aggiornata.
  15. In Ambiente, seleziona l'ambiente che vuoi utilizzare per eseguire il deployment dell'integrazione.
  16. Quando esegui il deployment del proxy API, fornisci l'email dell'account di servizio che hai creato in precedenza con il ruolo richiesto.
  17. Fai clic su Esegui il deployment.

Testare l'integrazione

Per testare l'integrazione, completa i seguenti passaggi:

  1. Scarica i dati di esempio in Cloud Shell:
    1. Per avviare una sessione di Cloud Shell dalla console Google Cloud, fai clic sul Icona Attiva Cloud Shell Attiva Cloud Shell nella Console Cloud: Viene avviata una nel riquadro inferiore della console Google Cloud.
    2. Inserisci il comando seguente nel terminale Cloud Shell: 
      wget https://raw.githubusercontent.com/GoogleCloudPlatform/application-integration-samples/main/assets/bq-sample-dataset.json
    3. Per verificare che i dati di esempio siano stati scaricati, inserisci il comando seguente nel terminale Cloud Shell: 
      ls -la bq-sample-dataset.json
       Il file scaricato viene elencato nel terminale Cloud Shell.
  2. Per abilitare il debug nel proxy Apigee:
    1. Torna al proxy API che hai appena creato nell'UI di Apigee.
    2. Fai clic sulla scheda Debug.
    3. Fai clic su Avvia sessione di debug e inserisci le seguenti informazioni:
      1. Seleziona l'ambiente in cui vuoi eseguire la sessione di debug.
      2. (Facoltativo) Nell'elenco a discesa Filtro, seleziona un filtro da applicare a tutte le transazioni nella sessione di debug che stai creando.Il valore predefinito è None (All transactions) e include tutte le transazioni nei dati di debug.
      3. Fai clic su Avvia.
  3. Per avviare il test, inserisci i comandi seguenti nel terminale Cloud Shell: 
    export APIGEE_DOMAIN=<your-Apigee-domain>
export SAMPLE_DOCS=$(jq  $(r=$((RANDOM % 1000)) ; echo ".[$r:$((r + 3))]") < bq-sample-dataset.json)
        
curl -X POST https://$APIGEE_DOMAIN/v1/process-records \
  -H 'Content-Type: application/json' \
  -d "$SAMPLE_DOCS"
    Questo test seleziona tre voci casuali dal set di dati campione e le passa all'integrazione principale. L'integrazione principale passa ogni voce all'integrazione secondaria, che aggiunge i dati come riga alla tabella BigQuery.
  4. Per verificare che la tabella BigQuery contenga ora questi record, segui questi passaggi:
    1. Nella pagina Cloud Console, fai clic su Menu di navigazione.
    2. Nella sezione Analisi, fai clic su BigQuery.
    3. Espandi il progetto e fai clic sul set di dati bq_tutorial.
    4. Espandi il set di dati bq_tutorial e fai clic sulla tabella tutorial.
    5. Fai clic sulla scheda Esplora tabella per visualizzare i record inseriti.

Passaggi successivi

Prova a creare integrazioni con altri connettori. Per l'elenco di tutti i connettori supportati vedi Riferimento del connettore.