Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Utilizza l'API Storage Write (REST)

L'API BigQuery Storage Write (REST), precedentemente nota come metodo tabledata.insertAll legacy, consente di inserire flussi di righe nelle tabelle BigQuery senza eseguire un job di caricamento. I dati in streaming diventano disponibili per l'analisi GoogleSQL in tempo reale non appena BigQuery conferma la richiesta di scrittura.

Per i nuovi progetti, utilizza l'API BigQuery Storage Write (gRPC) per prezzi più bassi, semantica di distribuzione exactly-once e flussi di dati nelle tabelle gestite di Apache Iceberg. Se stai eseguendo la migrazione di un progetto esistente dall'API Storage Write (REST) all'API Storage Write (gRPC), seleziona lo stream predefinito.

Prima di iniziare

Assicurati di disporre dell'accesso in scrittura al set di dati che contiene la tabella di destinazione. La tabella deve esistere prima di iniziare a scriverci dati, a meno che non utilizzi tabelle modello. Per saperne di più sulle tabelle modello, consulta Creazione automatica di tabelle utilizzando tabelle modello.
Consulta le norme relative alle quote per i dati in streaming.
Verifica che la fatturazione sia attivata per il tuo progetto Google Cloud .

Lo streaming non è disponibile tramite il Livello senza costi. Se tenti di utilizzare lo streaming senza attivare la fatturazione, riceverai il seguente errore: BigQuery: Streaming insert is not allowed in the free tier.

Concedi i ruoli IAM (Identity and Access Management) che forniscono agli utenti le autorizzazioni necessarie per eseguire ogni attività in questo documento.

Autorizzazioni obbligatorie

Per trasmettere flussi di dati in BigQuery, devi disporre delle seguenti autorizzazioni IAM:

bigquery.tables.updateData (consente di inserire dati nella tabella)
bigquery.tables.get (consente di ottenere i metadati della tabella)
bigquery.datasets.get (consente di ottenere i metadati del set di dati)
bigquery.tables.create (obbligatorio se utilizzi una tabella modello per creare automaticamente la tabella)

Ciascuno dei seguenti ruoli IAM predefiniti include le autorizzazioni necessarie per trasmettere dati in BigQuery:

roles/bigquery.dataEditor
roles/bigquery.dataOwner
roles/bigquery.admin

Per saperne di più sui ruoli e sulle autorizzazioni IAM in BigQuery, consulta Ruoli e autorizzazioni predefiniti.

Trasmetti flussi di dati in BigQuery

C#

Prima di provare questo esempio, segui le istruzioni di configurazione di C# nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery C#.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.


using Google.Cloud.BigQuery.V2;

public class BigQueryTableInsertRows
{
    public void TableInsertRows(
        string projectId = "your-project-id",
        string datasetId = "your_dataset_id",
        string tableId = "your_table_id"
    )
    {
        BigQueryClient client = BigQueryClient.Create(projectId);
        BigQueryInsertRow[] rows = new BigQueryInsertRow[]
        {
            // The insert ID is optional, but can avoid duplicate data
            // when retrying inserts.
            new BigQueryInsertRow(insertId: "row1") {
                { "name", "Washington" },
                { "post_abbr", "WA" }
            },
            new BigQueryInsertRow(insertId: "row2") {
                { "name", "Colorado" },
                { "post_abbr", "CO" }
            }
        };
        client.InsertRows(datasetId, tableId, rows);
    }
}

Go

Prima di provare questo esempio, segui le istruzioni di configurazione di Go nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Go.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

import (
	"context"
	"fmt"

	"cloud.google.com/go/bigquery"
)

// Item represents a row item.
type Item struct {
	Name string
	Age  int
}

// Save implements the ValueSaver interface.
// This example disables best-effort de-duplication, which allows for higher throughput.
func (i *Item) Save() (map[string]bigquery.Value, string, error) {
	return map[string]bigquery.Value{
		"full_name": i.Name,
		"age":       i.Age,
	}, bigquery.NoDedupeID, nil
}

// insertRows demonstrates inserting data into a table using the streaming insert mechanism.
func insertRows(projectID, datasetID, tableID string) error {
	// projectID := "my-project-id"
	// datasetID := "mydataset"
	// tableID := "mytable"
	ctx := context.Background()
	client, err := bigquery.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("bigquery.NewClient: %w", err)
	}
	defer client.Close()

	inserter := client.Dataset(datasetID).Table(tableID).Inserter()
	items := []*Item{
		// Item implements the ValueSaver interface.
		{Name: "Phred Phlyntstone", Age: 32},
		{Name: "Wylma Phlyntstone", Age: 29},
	}
	if err := inserter.Put(ctx, items); err != nil {
		return err
	}
	return nil
}

Java

Prima di provare questo esempio, segui le istruzioni di configurazione di Java nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Java.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to inserting rows into a table without running a load job.
public class TableInsertRows {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    // Create a row to insert
    Map<String, Object> rowContent = new HashMap<>();
    rowContent.put("booleanField", true);
    rowContent.put("numericField", "3.14");
    // TODO(developer): Replace the row id with a unique value for each row.
    String rowId = "ROW_ID";
    tableInsertRows(datasetName, tableName, rowId, rowContent);
  }

  public static void tableInsertRows(
      String datasetName, String tableName, String rowId, Map<String, Object> rowContent) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

      // Get table
      TableId tableId = TableId.of(datasetName, tableName);

      // Inserts rowContent into datasetName:tableId.
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(tableId)
                  // More rows can be added in the same RPC by invoking .addRow() on the builder.
                  // You can omit the unique row ids to disable de-duplication.
                  .addRow(rowId, rowContent)
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

Node.js

Prima di provare questo esempio, segui le istruzioni di configurazione di Node.js nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Node.js.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

// Import the Google Cloud client library
const {BigQuery} = require('@google-cloud/bigquery');
const bigquery = new BigQuery();

async function insertRowsAsStream() {
  // Inserts the JSON objects into my_dataset:my_table.

  /**
   * TODO(developer): Uncomment the following lines before running the sample.
   */
  // const datasetId = 'my_dataset';
  // const tableId = 'my_table';
  const rows = [
    {name: 'Tom', age: 30},
    {name: 'Jane', age: 32},
  ];

  // Insert data into a table
  await bigquery.dataset(datasetId).table(tableId).insert(rows);
  console.log(`Inserted ${rows.length} rows`);
}

PHP

Prima di provare questo esempio, segui le istruzioni di configurazione di PHP nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery PHP.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

use Google\Cloud\BigQuery\BigQueryClient;

/**
 * Stream data into bigquery
 *
 * @param string $projectId The project Id of your Google Cloud Project.
 * @param string $datasetId The BigQuery dataset ID.
 * @param string $tableId The BigQuery table ID.
 * @param string $data Json encoded data For eg,
 *    $data = json_encode([
 *       "field1" => "value1",
 *       "field2" => "value2",
 *    ]);
 */
function stream_row(
    string $projectId,
    string $datasetId,
    string $tableId,
    string $data
): void {
    // instantiate the bigquery table service
    $bigQuery = new BigQueryClient([
      'projectId' => $projectId,
    ]);
    $dataset = $bigQuery->dataset($datasetId);
    $table = $dataset->table($tableId);

    $data = json_decode($data, true);
    $insertResponse = $table->insertRows([
      ['data' => $data],
      // additional rows can go here
    ]);
    if ($insertResponse->isSuccessful()) {
        print('Data streamed into BigQuery successfully' . PHP_EOL);
    } else {
        foreach ($insertResponse->failedRows() as $row) {
            foreach ($row['errors'] as $error) {
                printf('%s: %s' . PHP_EOL, $error['reason'], $error['message']);
            }
        }
    }
}

Python

Prima di provare questo esempio, segui le istruzioni di configurazione di Python nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Python.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

from google.cloud import bigquery

# Construct a BigQuery client object.
client = bigquery.Client()

# TODO(developer): Set table_id to the ID of table to append to.
# table_id = "your-project.your_dataset.your_table"

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(table_id, rows_to_insert)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

Ruby

Prima di provare questo esempio, segui le istruzioni di configurazione di Ruby nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Ruby.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

require "google/cloud/bigquery"

def table_insert_rows dataset_id = "your_dataset_id", table_id = "your_table_id"
  bigquery = Google::Cloud::Bigquery.new
  dataset  = bigquery.dataset dataset_id
  table    = dataset.table table_id

  row_data = [
    { name: "Alice", value: 5  },
    { name: "Bob",   value: 10 }
  ]
  response = table.insert row_data

  if response.success?
    puts "Inserted rows successfully"
  else
    puts "Failed to insert #{response.error_rows.count} rows"
  end
end

Non è necessario compilare il campo insertID quando inserisci le righe. L'esempio seguente mostra come evitare di inviare un insertID per ogni riga durante lo streaming.

Java

Prima di provare questo esempio, segui le istruzioni di configurazione di Java nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Java.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import com.google.common.collect.ImmutableList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to insert rows without row ids in a table
public class TableInsertRowsWithoutRowIds {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    tableInsertRowsWithoutRowIds(datasetName, tableName);
  }

  public static void tableInsertRowsWithoutRowIds(String datasetName, String tableName) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      final BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
      // Create rows to insert
      Map<String, Object> rowContent1 = new HashMap<>();
      rowContent1.put("stringField", "Phred Phlyntstone");
      rowContent1.put("numericField", 32);
      Map<String, Object> rowContent2 = new HashMap<>();
      rowContent2.put("stringField", "Wylma Phlyntstone");
      rowContent2.put("numericField", 29);
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(TableId.of(datasetName, tableName))
                  // No row ids disable de-duplication, and also disable the retries in the Java
                  // library.
                  .setRows(
                      ImmutableList.of(
                          InsertAllRequest.RowToInsert.of(rowContent1),
                          InsertAllRequest.RowToInsert.of(rowContent2)))
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table without row ids");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

Python

Prima di provare questo esempio, segui le istruzioni di configurazione di Python nella guida rapida di BigQuery per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API BigQuery Python.

Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.

from google.cloud import bigquery

# Construct a BigQuery client object.
client = bigquery.Client()

# TODO(developer): Set table_id to the ID of table to append to.
# table_id = "your-project.your_dataset.your_table"

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(
    table_id, rows_to_insert, row_ids=[None] * len(rows_to_insert)
)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

Invia dati relativi a data e ora

Per i campi di data e ora, formatta i dati nell'API Storage Write (REST) come segue:

Tipo	Formato
`DATE`	Una stringa nel formato `"YYYY-MM-DD"`
`DATETIME`	Una stringa nel formato `"YYYY-MM-DD [HH:MM:SS]"`
`TIME`	Una stringa nel formato `"HH:MM:SS"`
`TIMESTAMP`	Il numero di secondi trascorsi dal 1° gennaio 1970 (epoca di Unix) o una stringa nel formato `"YYYY-MM-DD HH:MM[:SS]"`

Invia dati dell'intervallo

Per i campi con tipo RANGE<T>, formatta i dati nell'API Storage Write (REST) come oggetto JSON con due campi, start e end. I valori mancanti o NULL per i campi start e end rappresentano limiti non vincolati. Questi campi devono avere lo stesso formato JSON supportato di tipo T, dove T può essere uno tra DATE, DATETIME e TIMESTAMP.

Nell'esempio seguente, il campo f_range_date rappresenta una colonna RANGE<DATE> in una tabella. Una riga viene inserita in questa colonna utilizzando l'API Storage Write (REST).

{
    "f_range_date": {
        "start": "1970-01-02",
        "end": null
    }
}

Disponibilità dei dati dello stream

I dati sono disponibili per l'analisi in tempo reale utilizzando le query GoogleSQL immediatamente dopo che BigQuery ha confermato correttamente una richiesta dell'API Storage Write (REST). Quando esegui query sui dati nel buffer di flussi di dati, non ti vengono addebitati i byte elaborati dal buffer di flussi di dati se utilizzi i prezzi di computing on demand. Se utilizzi i prezzi basati sulla capacità, le tue prenotazioni consumano slot per l'elaborazione dei dati nel buffer di flussi di dati.

Le righe trasmesse di recente in una tabella partizionata per data di acquisizione hanno temporaneamente un valore NULL per la pseudocolonna _PARTITIONTIME. Per queste righe, BigQuery assegna il valore finale non NULL della colonna PARTITIONTIME in background, in genere entro pochi minuti. In rari casi, l'operazione può richiedere fino a 90 minuti.

Alcune righe trasmesse in streaming di recente potrebbero non essere disponibili per la copia della tabella, in genere per alcuni minuti. In rari casi, l'operazione può richiedere fino a 90 minuti. Per verificare se i dati sono disponibili per la copia della tabella, controlla la risposta tables.get per una sezione denominata streamingBuffer. Se la sezione streamingBuffer non è presente, i tuoi dati sono disponibili per la copia. Puoi anche utilizzare il campo streamingBuffer.oldestEntryTime per identificare l'età dei record nel buffer di flussi di dati.

Deduplicazione best effort

Quando fornisci insertId per una riga inserita, BigQuery utilizza questo ID per supportare la deduplicazione best-effort per un massimo di un minuto. In altre parole, se esegui lo streaming della stessa riga con lo stesso insertId più di una volta entro questo periodo di tempo nella stessa tabella, BigQuery potrebbe deduplicare le occorrenze multiple di quella riga, conservandone solo una.

Il sistema si aspetta che le righe fornite con insertId identici siano anch'esse identiche. Se due righe hanno insertId identici, non è deterministico quale riga BigQuery conserva.

La deduplicazione è generalmente destinata a scenari di nuovi tentativi in un sistema distribuito in cui non è possibile determinare lo stato di un inserimento di flussi di dati in determinate condizioni di errore, ad esempio errori di rete tra il tuo sistema e BigQuery o errori interni in BigQuery. Se riprovi un inserimento, utilizza lo stesso insertId per lo stesso insieme di righe in modo che BigQuery possa tentare di deduplicare i dati. Per saperne di più, consulta la sezione Risoluzione dei problemi relativi agli inserimenti di flussi di dati.

La deduplicazione offerta da BigQuery è un'operazione best effort e non deve essere considerata un meccanismo per garantire l'assenza di duplicati nei dati. Inoltre, BigQuery potrebbe ridurre la qualità della deduplicazione best effort in qualsiasi momento per garantire una maggiore affidabilità e disponibilità dei dati.

Se hai requisiti rigorosi di deduplicazione per i tuoi dati, Google Cloud Datastore è un servizio alternativo che supporta le transazioni.

Disattivazione della deduplicazione best effort

Puoi disattivare la deduplicazione best effort non compilando il campo insertId per ogni riga inserita. Questo è il modo consigliato per inserire i dati.

Apache Beam e Dataflow

Per disattivare la deduplicazione best effort quando utilizzi il connettore BigQuery I/O di Apache Beam per Java, utilizza il metodo ignoreInsertIds().

Rimozione manuale dei duplicati

Per assicurarti che non esistano righe duplicate al termine dello streaming, utilizza la seguente procedura manuale:

Aggiungi insertId come colonna nello schema della tabella e includi il valore insertId nei dati di ogni riga.

Al termine dello streaming, esegui la seguente query per verificare la presenza di duplicati:

#standardSQL
SELECT
  MAX(count) FROM(
  SELECT
    ID_COLUMN,
    count(*) as count
  FROM
    `TABLE_NAME`
  GROUP BY
    ID_COLUMN)

Se il risultato è maggiore di 1, esistono duplicati.

Per rimuovere i duplicati, esegui questa query. Specifica una tabella di destinazione, consenti risultati di grandi dimensioni e disabilita l'appiattimento dei risultati.

#standardSQL
SELECT
  * EXCEPT(row_number)
FROM (
  SELECT
    *,
    ROW_NUMBER()
          OVER (PARTITION BY ID_COLUMN) row_number
  FROM
    `TABLE_NAME`)
WHERE
  row_number = 1

Note sulla query di rimozione dei duplicati:

La strategia più sicura per la query di rimozione dei duplicati è scegliere come target una nuova tabella. In alternativa, puoi scegliere come target la tabella di origine con la disposizione di scrittura WRITE_TRUNCATE.
La query di rimozione dei duplicati aggiunge una colonna row_number con il valore 1 alla fine dello schema della tabella. La query utilizza un'istruzione SELECT * EXCEPT di GoogleSQL per escludere la colonna row_number dalla tabella di destinazione. Il prefisso #standardSQL attiva GoogleSQL per questa query. In alternativa, puoi selezionare nomi di colonne specifici per omettere questa colonna.
Per eseguire query su dati in tempo reale con i duplicati rimossi, puoi anche creare una visualizzazione della tabella utilizzando la query di rimozione dei duplicati. Tieni presente che i costi delle query sulla vista vengono calcolati in base alle colonne selezionate nella vista, il che può comportare dimensioni elevate dei byte analizzati.

Esegui lo streaming in tabelle partizionate nel tempo

Quando trasmetti dati in streaming a una tabella partizionata per tempo, ogni partizione ha un buffer di streaming. Il buffer di flussi di dati viene conservato quando esegui un job di caricamento, query o copia che sovrascrive una partizione impostando la proprietà writeDisposition su WRITE_TRUNCATE. Se vuoi rimuovere il buffer di streaming, verifica che sia vuoto chiamando tables.get sulla partizione.

Partizionamento per data di importazione

Quando esegui lo streaming in una tabella partizionata per data di importazione, BigQuery deduce la partizione di destinazione dall'ora UTC corrente.

I dati appena arrivati vengono inseriti temporaneamente nella partizione __UNPARTITIONED__ mentre si trovano nel buffer di flussi di dati. Quando i dati non partizionati sono sufficienti, BigQuery li suddivide nella partizione corretta. Tuttavia, non esiste uno SLA per il tempo necessario per lo spostamento dei dati dalla partizione __UNPARTITIONED__. Una query può escludere i dati nel buffer di flussi di dati da una query filtrando i valori NULL dalla partizione __UNPARTITIONED__ utilizzando una delle pseudocolonne (_PARTITIONTIME o _PARTITIONDATE a seconda del tipo di dati preferito).

Se stai trasmettendo dati in una tabella partizionata giornalmente, puoi eseguire l'override dell'inferenza della data fornendo un decoratore di partizione nell'ambito dell'API Storage Write (REST). Includi il decoratore nel parametro tableId. Ad esempio, puoi eseguire lo streaming nella partizione corrispondente al 01/03/2021 per la tabella table1 utilizzando l'indicatore di partizione:

table1$20210301

Quando utilizzi un decoratore di partizione, puoi trasmettere in streaming alle partizioni negli ultimi 31 giorni nel passato e nei 16 giorni nel futuro rispetto alla data corrente, in base all'ora UTC corrente. Per scrivere in partizioni per date al di fuori di questi limiti consentiti, utilizza un job di caricamento o di query, come descritto in Aggiunta e sovrascrittura dei dati delle tabella partizionata partizionate.

Lo streaming con un decoratore di partizione è supportato solo per le tabelle partizionate giornalmente. Non è supportato per le tabelle partizionate per ora, mese o anno.

Per i test, puoi utilizzare il comando dell'interfaccia a riga di comando bq insert dello strumento a riga di comando bq. Ad esempio, il seguente comando trasmette in streaming una singola riga a una partizione per la data 1° gennaio 2017 ($20170101) in una tabella partizionata denominata mydataset.mytable:

echo '{"a":1, "b":2}' | bq insert 'mydataset.mytable$20170101'

Partizionamento delle colonne per unità di tempo

Puoi trasmettere dati in una tabella partizionata in base a una colonna DATE, DATETIME o TIMESTAMP compresa tra 10 anni nel passato e 1 anno nel futuro. I dati non compresi nell'intervallo vengono rifiutati.

Quando i dati vengono trasmessi in streaming, vengono inizialmente inseriti nella partizione __UNPARTITIONED__. Quando i dati non partizionati sono sufficienti, BigQuery li ripartiziona automaticamente, inserendoli nella partizione appropriata. Tuttavia, non esiste uno SLA per il tempo necessario per lo spostamento dei dati dalla partizione __UNPARTITIONED__.

Nota: le partizioni giornaliere vengono elaborate in modo diverso rispetto alle partizioni orarie, mensili e annuali. Solo i dati al di fuori dell'intervallo di date (dagli ultimi 7 giorni ai prossimi 3 giorni) vengono estratti nella partizione UNPARTITIONED, in attesa di essere ripartizionati. Invece, per la tabella partizionata per ora, i dati vengono sempre estratti nella partizione UNPARTITIONED e poi partizionati di nuovo.

Crea tabelle automaticamente utilizzando tabelle modello

Le tabelle modello forniscono un meccanismo per suddividere una tabella logica in molte tabelle più piccole per creare set di dati più piccoli (ad esempio, per ID utente). Le tabelle modello presentano una serie di limitazioni descritte più avanti in questa sezione. Per ottenere questo comportamento, è consigliabile utilizzare invece le tabelle partizionate e le tabelle in cluster.

Per utilizzare una tabella modello tramite l'API BigQuery, aggiungi un parametro templateSuffix alla richiesta dell'API Storage Write (REST). Per lo strumento a riga di comando bq, aggiungi il flag template_suffix al comando insert. Se BigQuery rileva un parametro templateSuffix o il flag template_suffix, tratta la tabella di destinazione come un modello di base. Crea una nuova tabella che condivide lo stesso schema della tabella di destinazione e ha un nome che include il suffisso specificato:

<targeted_table_name> + <templateSuffix>

Utilizzando una tabella modello, eviti l'overhead della creazione di ogni tabella singolarmente e della specifica dello schema per ogni tabella. Devi solo creare un singolo modello e fornire suffissi diversi in modo che BigQuery possa creare le nuove tabelle per te. BigQuery inserisce le tabelle nello stesso progetto e nello stesso set di dati.

Le tabelle create utilizzando le tabelle modello sono in genere disponibili entro pochi secondi. In rari casi, potrebbe essere necessario più tempo prima che diventino disponibili.

Modifica lo schema della tabella del modello

Se modifichi lo schema di una tabella modello, tutte le tabelle generate successivamente utilizzano lo schema aggiornato. Le tabelle generate in precedenza non sono interessate, a meno che la tabella esistente non abbia ancora un buffer di flussi.

Per le tabelle esistenti che hanno ancora un buffer di flussi di dati, se modifichi lo schema della tabella modello in modo compatibile con le versioni precedenti, viene aggiornato anche lo schema delle tabelle generate con flussi di dati attivi. Tuttavia, se modifichi lo schema della tabella modello in modo non compatibile con le versioni precedenti, tutti i dati memorizzati nel buffer che utilizzano lo schema precedente andranno persi. Inoltre, non puoi trasmettere nuovi dati alle tabelle generate esistenti che utilizzano lo schema precedente, ma ora incompatibile.

Dopo aver modificato lo schema di una tabella modello, attendi che le modifiche vengano propagate prima di provare a inserire nuovi dati o eseguire query sulle tabelle generate. Le richieste di inserimento di nuovi campi dovrebbero essere completate entro pochi minuti. I tentativi di eseguire query sui nuovi campi potrebbero richiedere un'attesa più lunga, fino a 90 minuti.

Se vuoi modificare lo schema di una tabella generata, non modificarlo finché non viene interrotto lo streaming tramite la tabella modello e la sezione delle statistiche di streaming della tabella generata non è più presente nella risposta tables.get(), il che indica che non sono presenti dati nel buffer della tabella.

Le tabelle partizionate e le tabelle in cluster non presentano le limitazioni precedenti e sono il meccanismo consigliato.

Dettagli della tabella del modello

Valore del suffisso del modello: Il valore templateSuffix (o --template_suffix) deve contenere solo lettere (a-z, A-Z), numeri (0-9) o trattini bassi (_). La lunghezza massima combinata del nome della tabella e del suffisso della tabella è di 1024 caratteri.
Quota: Le tabelle modello sono soggette alle limitazioni della quota di streaming. Il tuo progetto può creare fino a 10 tabelle al secondo con le tabelle modello, in modo simile all'API tables.insert. Questa quota si applica solo alle tabelle in fase di creazione, non a quelle in fase di modifica.; Se la tua applicazione deve creare più di 10 tabelle al secondo, ti consigliamo di utilizzare tabelle in cluster. Ad esempio, puoi inserire l'ID della tabella ad alta cardinalità nella colonna chiave di una singola tabella in cluster.
Durata: La tabella generata eredita la scadenza dal set di dati. Come per i normali dati in streaming, le tabelle generate non possono essere copiate immediatamente.
Deduplicazione: La deduplicazione avviene solo tra riferimenti uniformi a una tabella di destinazione. Ad esempio, se trasmetti in streaming contemporaneamente a una tabella generata utilizzando sia tabelle modello sia un comando API Storage Write (REST) standard, non viene eseguita alcuna deduplicazione tra le righe inserite dalle tabelle modello e un comando API Storage Write (REST) standard.
Visualizzazioni: La tabella modello e le tabelle generate non devono essere viste.

Risolvi i problemi relativi agli inserimenti di flussi di dati

Le sezioni seguenti spiegano come risolvere gli errori che si verificano quando trasmetti dati in BigQuery utilizzando l'API Storage Write (REST). Per saperne di più su come risolvere gli errori relativi alla quota per gli inserimenti di flussi di dati, consulta Errori relativi alla quota per gli inserimenti di flussi di dati.

Codici di risposta HTTP di errore

Se ricevi un codice di risposta HTTP di errore, ad esempio un errore di rete, non è possibile stabilire se l'inserimento di flussi di dati è andato a buon fine. Se provi a inviare nuovamente la richiesta, potresti ottenere righe duplicate nella tabella. Per proteggere la tabella dalla duplicazione, imposta la proprietà insertId quando invii la richiesta. BigQuery utilizza la proprietà insertId per la deduplicazione.

Se ricevi un errore di autorizzazione, un errore di nome della tabella non valido o un errore di superamento della quota, non vengono inserite righe e l'intera richiesta non va a buon fine.

Codici di risposta HTTP di esito positivo

Anche se ricevi un codice di risposta HTTP di operazione riuscita, devi controllare la proprietà insertErrors della risposta per determinare se gli inserimenti di righe sono andati a buon fine, perché BigQuery potrebbe riuscire solo parzialmente a inserire le righe. Potresti riscontrare uno dei seguenti scenari:

Tutte le righe sono state inserite correttamente: se la proprietà insertErrors è un elenco vuoto, tutte le righe sono state inserite correttamente.
Alcune righe sono state inserite correttamente: ad eccezione dei casi in cui esiste una mancata corrispondenza dello schema in una delle righe, le righe indicate nella proprietà insertErrors non vengono inserite e tutte le altre righe vengono inserite correttamente. La proprietà errors contiene informazioni dettagliate sul motivo per cui ogni riga non è riuscita. La proprietà index indica l'indice di riga in base zero della richiesta a cui si applica l'errore.
Nessuna riga inserita correttamente: se BigQuery rileva una mancata corrispondenza dello schema su singole righe della richiesta, non viene inserita nessuna riga e viene restituita una voce insertErrors per ogni riga, anche per quelle che non presentavano una mancata corrispondenza dello schema. Le righe che non presentavano una mancata corrispondenza dello schema presentano un errore con la proprietà reason impostata su stopped e puoi inviarle di nuovo così come sono. Le righe che non sono state inviate includono informazioni dettagliate sulla mancata corrispondenza dello schema. Per scoprire i tipi di buffer di protocollo supportati per ogni tipo di dati BigQuery, consulta Tipi di dati Arrow e buffer di protocollo supportati.

Errori dei metadati per gli inserimenti di flussi di dati

Poiché l'API BigQuery Streaming è progettata per frequenze di inserimento elevate, le modifiche ai metadati della tabella sottostante sono a coerenza finale quando interagiscono con il sistema di streaming. La maggior parte delle volte, le modifiche ai metadati vengono propagate entro pochi minuti, ma durante questo periodo le risposte dell'API potrebbero riflettere lo stato incoerente della tabella.

Alcuni scenari includono:

Modifiche allo schema: la modifica dello schema di una tabella che ha ricevuto di recente inserimenti di flussi di dati può causare risposte con errori di mancata corrispondenza dello schema, perché il sistema di flussi di dati potrebbe non rilevare immediatamente la modifica dello schema.
Creazione o eliminazione di tabelle: lo streaming in una tabella inesistente restituisce una variante di una risposta notFound. Una tabella creata in risposta potrebbe non essere riconosciuta immediatamente dagli inserimenti di flussi di dati successivi. Analogamente, l'eliminazione o la ricreazione di una tabella può creare un periodo di tempo in cui gli inserimenti di flussi di dati vengono inviati alla vecchia tabella. Gli inserimenti di flussi di dati potrebbero non essere presenti nella nuova tabella.
Troncamento della tabella: il troncamento dei dati di una tabella (utilizzando un job di query che utilizza un valore writeDisposition di WRITE_TRUNCATE) può causare l'eliminazione degli inserimenti successivi durante il periodo di coerenza.

Dati mancanti o non disponibili

Gli inserimenti di flussi di dati si trovano temporaneamente nello spazio di archiviazione ottimizzato per la scrittura, che ha caratteristiche di disponibilità diverse rispetto allo spazio di archiviazione gestito. Alcune operazioni in BigQuery non interagiscono con lo spazio di archiviazione ottimizzato per la scrittura, ad esempio i job di copia delle tabelle e i metodi API come tabledata.list. I dati di streaming recenti non sono presenti nella tabella di destinazione o nell'output.

Errori di quota relativi agli inserimenti di flussi di dati

Questa sezione fornisce suggerimenti per la risoluzione dei problemi relativi agli errori di quota per lo streaming di dati in BigQuery.

In alcune aree geografiche, gli inserimenti di flussi di dati hanno una quota maggiore se non immetti dati nel campo insertId per ogni riga. Per ulteriori informazioni sulle quote per gli inserimenti di flussi di dati, consulta la pagina relativa agli Inserimento di flussi di dati. Gli errori relativi alle quote per i flussi di dati di BigQuery dipendono dalla presenza o dall'assenza di un valore nel campo insertId.

Messaggio di errore

Se il campo insertId è vuoto, si può verificare il seguente errore di quota:

Limite quota	Messaggio di errore
Byte al secondo per progetto	La tua entità con ID GAIA `GAIA_ID`, progetto `PROJECT_ID` e area geografica `REGION` ha superato la quota di inserimento byte al secondo.

Se il campo insertId è compilato, si possono verificare i seguenti errori di quota:

Limite quota	Messaggio di errore
Righe al secondo per progetto	Il tuo progetto `PROJECT_ID` in `REGION` ha superato la quota per l'inserimento di righe di flussi di dati al secondo.
Righe al secondo per tabella	La tua tabella: `TABLE_ID` ha superato la quota per le righe di inserimento di flussi di dati al secondo.
Byte al secondo per tabella	La tua tabella: `TABLE_ID` ha superato la quota per i byte al secondo di inserimento di flussi di dati.

Lo scopo del campo insertId è deduplicare le righe inserite. Se arrivano più inserimenti con lo stesso insertId nel giro di pochi minuti, BigQuery scrive un'unica versione del record. Tuttavia, questa deduplicazione automatica non è garantita. Per la massima velocità effettiva di trasmissione dei flussi di dati, ti consigliamo di non includere insertId e di usare invece la deduplicazione manuale. Per ulteriori informazioni, consulta la pagina relativa a come garantire la coerenza dei dati.

Quando si verifica questo errore, diagnostica il problema e poi segui i passaggi consigliati per risolverlo.

Diagnosi

Usa le viste STREAMING_TIMELINE_BY_* per analizzare il traffico dei flussi di dati. Queste visualizzazioni aggregano le statistiche di streaming su intervalli di un minuto, raggruppati per error_code. Gli errori di quota vengono visualizzati nei risultati con error_code uguale a RATE_LIMIT_EXCEEDED o QUOTA_EXCEEDED.

A seconda del limite di quota specifico raggiunto, fai riferimento a total_rows o total_input_bytes. Se l'errore riguarda una quota a livello di tabella, filtra per table_id.

Ad esempio, la seguente query mostra i byte totali importati al minuto e il numero totale di errori di quota:

SELECT
 start_timestamp,
 error_code,
 SUM(total_input_bytes) as sum_input_bytes,
 SUM(IF(error_code IN ('QUOTA_EXCEEDED', 'RATE_LIMIT_EXCEEDED'),
     total_requests, 0)) AS quota_error
FROM
 `region-REGION_NAME`.INFORMATION_SCHEMA.STREAMING_TIMELINE_BY_PROJECT
WHERE
  start_timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP, INTERVAL 1 DAY)
GROUP BY
 start_timestamp,
 error_code
ORDER BY 1 DESC

Risoluzione

Per risolvere questo errore di quota, segui questi passaggi:

Se utilizzi il campo insertId per la deduplicazione e il tuo progetto si trova in una regione che supporta la quota di flussi di dati più elevata, ti consigliamo di rimuovere il campo insertId. Questa soluzione potrebbe richiedere alcuni passaggi aggiuntivi per deduplicare manualmente i dati. Per ulteriori informazioni, vedi Rimozione manuale dei duplicati.
Se non usi insertId, oppure non è possibile rimuoverlo, monitora il traffico dei flussi di dati per un periodo di 24 ore e analizza gli errori di quota:
- Se visualizzi principalmente errori RATE_LIMIT_EXCEEDED anziché errori QUOTA_EXCEEDED e il traffico complessivo è inferiore all'80% della quota, probabilmente gli errori indicano picchi temporanei. Puoi risolvere questi errori riprovando l'operazione utilizzando il backoff esponenziale tra i nuovi tentativi.
- Se utilizzi un job Dataflow per inserire dati, valuta la possibilità di utilizzare job di caricamento anziché inserimenti di flussi di dati. Per saperne di più, consulta Impostazione del metodo di inserimento. Se utilizzi Dataflow con un connettore I/O personalizzato, valuta la possibilità di utilizzare un connettore I/O integrato. Per saperne di più, consulta Pattern di I/O personalizzati.
- Se visualizzi errori QUOTA_EXCEEDED o se il traffico complessivo supera costantemente l'80% della quota, invia una richiesta di aumento della quota. Per saperne di più, consulta Richiedi un aggiustamento delle quote.
- Potresti anche prendere in considerazione la possibilità di sostituire gli inserimenti di flussi di dati con la più recente API Storage Write, che offre un throughput più elevato, un prezzo inferiore e molte funzionalità utili.

Salvo quando diversamente specificato, i contenuti di questa pagina sono concessi in base alla licenza Creative Commons Attribution 4.0, mentre gli esempi di codice sono concessi in base alla licenza Apache 2.0. Per ulteriori dettagli, consulta le norme del sito di Google Developers. Java è un marchio registrato di Oracle e/o delle sue consociate.

Ultimo aggiornamento 2026-10-30 UTC.