Trasmetti i dati in streaming utilizzando l'API Storage Write (gRPC)
L'API BigQuery Storage Write (gRPC) consente di trasmettere continuamente flussi di dati alle tabelle BigQuery in modo che i record siano disponibili per le query con una latenza minima. A seconda dei requisiti del workload, puoi utilizzare il flusso predefinito per le scritture "at-least-once" o creare flussi di tipo "committed" con offset del flusso per la distribuzione "exactly-once".
Utilizza lo stream predefinito per la semantica di tipo "at-least-once"
Se la tua applicazione può accettare la possibilità che nella tabella di destinazione vengano visualizzati record duplicati, ti consigliamo di utilizzare lo stream predefinito per gli scenari di streaming. Lo stream predefinito utilizza il tipo committed, quindi i record sono disponibili per le query non appena il server conferma la richiesta di scrittura, senza distribuzione exactly-once.
Il seguente codice mostra come scrivere dati nello stream predefinito:
Java
Per scoprire come installare e utilizzare la libreria client per BigQuery, consulta Librerie client di BigQuery. 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.
Node.js
Per scoprire come installare e utilizzare la libreria client per BigQuery, consulta Librerie client di BigQuery.
Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.
Python
Questo esempio mostra come inserire un record con due campi utilizzando lo stream predefinito:
from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types
from google.cloud.bigquery_storage_v1 import writer
from google.protobuf import descriptor_pb2
import logging
import json
import sample_data_pb2
# The list of columns from the table's schema to search in the given data to write to BigQuery.
TABLE_COLUMNS_TO_CHECK = [
"name",
"age"
]
# Function to create a batch of row data to be serialized.
def create_row_data(data):
row = sample_data_pb2.SampleData()
for field in TABLE_COLUMNS_TO_CHECK:
# Optional fields will be passed as null if not provided
if field in data:
setattr(row, field, data[field])
return row.SerializeToString()
class BigQueryStorageWriteAppend(object):
# The stream name is: projects/{project}/datasets/{dataset}/tables/{table}/_default
def append_rows_proto2(
project_id: str, dataset_id: str, table_id: str, data: dict
):
write_client = bigquery_storage_v1.BigQueryWriteClient()
parent = write_client.table_path(project_id, dataset_id, table_id)
stream_name = f'{parent}/_default'
write_stream = types.WriteStream()
# Create a template with fields needed for the first request.
request_template = types.AppendRowsRequest()
# The request must contain the stream name.
request_template.write_stream = stream_name
# Generating the protocol buffer representation of the message descriptor.
proto_schema = types.ProtoSchema()
proto_descriptor = descriptor_pb2.DescriptorProto()
sample_data_pb2.SampleData.DESCRIPTOR.CopyToProto(proto_descriptor)
proto_schema.proto_descriptor = proto_descriptor
proto_data = types.AppendRowsRequest.ProtoData()
proto_data.writer_schema = proto_schema
request_template.proto_rows = proto_data
# Construct an AppendRowsStream to send an arbitrary number of requests to a stream.
append_rows_stream = writer.AppendRowsStream(write_client, request_template)
# Append proto2 serialized bytes to the serialized_rows repeated field using create_row_data.
proto_rows = types.ProtoRows()
for row in data:
proto_rows.serialized_rows.append(create_row_data(row))
# Appends data to the given stream.
request = types.AppendRowsRequest()
proto_data = types.AppendRowsRequest.ProtoData()
proto_data.rows = proto_rows
request.proto_rows = proto_data
append_rows_stream.send(request)
print(f"Rows to table: '{parent}' have been written.")
if __name__ == "__main__":
###### Uncomment the below block to provide additional logging capabilities ######
#logging.basicConfig(
# level=logging.DEBUG,
# format="%(asctime)s [%(levelname)s] %(message)s",
# handlers=[
# logging.StreamHandler()
# ]
#)
###### Uncomment the above block to provide additional logging capabilities ######
with open('entries.json', 'r') as json_file:
data = json.load(json_file)
# Change this to your specific BigQuery project, dataset, table details
BigQueryStorageWriteAppend.append_rows_proto2("PROJECT_ID","DATASET_ID", "TABLE_ID ",data=data)
Questo esempio di codice dipende dal modulo di protocollo compilato sample_data_pb2.py. Per creare il modulo compilato, esegui il comando protoc --python_out=. sample_data.proto, dove protoc è il compilatore del buffer di protocollo. Il file sample_data.proto definisce il formato dei messaggi utilizzati nell'esempio Python. Per installare il compilatore protoc, segui le istruzioni riportate in Protocol Buffers - Google's data interchange format.
Ecco i contenuti del file sample_data.proto:
message SampleData {
required string name = 1;
required int64 age = 2;
}
Questo script utilizza il file entries.json, che contiene dati di riga di esempio da inserire nella tabella BigQuery:
{"name": "Jim", "age": 35}
{"name": "Jane", "age": 27}
Rust
Utilizza il multiplexing
Puoi attivare il multiplexing a livello di writer di stream solo per lo stream predefinito. Per attivare il multiplexing in Java, chiama il metodo setEnableConnectionPool quando crei un oggetto StreamWriter o JsonStreamWriter.
Dopo aver abilitato il pool di connessioni, la libreria client Java gestisce le connessioni in background, aumentandole se quelle esistenti sono considerate troppo occupate. Per rendere più efficace la scalabilità automatica, ti consigliiamo di ridurre il limite maxInflightRequests.
// One possible way for constructing StreamWriter StreamWriter.newBuilder(streamName) .setWriterSchema(protoSchema) .setEnableConnectionPool(true) .setMaxInflightRequests(100) .build(); // One possible way for constructing JsonStreamWriter JsonStreamWriter.newBuilder(tableName, bigqueryClient) .setEnableConnectionPool(true) .setMaxInflightRequests(100) .build();
Per abilitare il multiplexing in Go, consulta Condivisione della connessione (multiplexing).
Per attivare il multiplexing in Rust, chiama il metodo with_multiplexing(true) quando crei un writer di stream predefinito:
let writer = client .open_default_stream(table_name) .with_multiplexing(true) .build_arrow(schema) .await?;
Utilizza il tipo di commit per la semantica "exactly-once"
Se hai bisogno di una semantica di scrittura exactly-once, crea un flusso di scrittura di tipo committed. Nel tipo committed, i record sono disponibili per la query non appena il client riceve la conferma dal backend.
Il tipo Committed fornisce la consegna "exactly-once" all'interno di uno stream tramite l'utilizzo di offset dei record. Utilizzando gli offset dei record, l'applicazione specifica l'offset di aggiunta successivo in ogni chiamata a AppendRows. L'operazione di scrittura viene eseguita solo se il valore di offset corrisponde all'offset di aggiunta successivo. Per saperne di più, consulta Gestisci gli offset degli stream per ottenere la semantica exactly-once.
Se non fornisci un offset, i record vengono aggiunti alla fine corrente dello stream. In questo caso, se una richiesta di aggiunta restituisce un errore, riprovare potrebbe comportare la visualizzazione del record più di una volta nello stream.
Per utilizzare il tipo di commit, segui questi passaggi:
Java
- Chiama
CreateWriteStreamper creare uno o più stream nel tipo committed. - Per ogni stream, chiama
AppendRowsin un loop per scrivere batch di record. - Chiama
FinalizeWriteStreamper ogni stream per rilasciarlo. Dopo aver chiamato questo metodo, non puoi scrivere altre righe nello stream. Questo passaggio è facoltativo nel tipo impegnato, ma aiuta a evitare di superare il limite di stream attivi. Per saperne di più, consulta Limita la frequenza di creazione degli stream.
Node.js
- Chiama
createWriteStreamFullResponseper creare uno o più stream nel tipo committed. - Per ogni stream, chiama
appendRowsin un loop per scrivere batch di record. - Chiama
finalizeper ogni stream per rilasciarlo. Dopo aver chiamato questo metodo, non puoi scrivere altre righe nello stream. Questo passaggio è facoltativo nel tipo committed, ma aiuta a evitare di superare il limite di stream attivi. Per saperne di più, consulta Limita la frequenza di creazione degli stream.
Rust
- Chiama
create_streamsul clientWritee crea unCommittedWriterper creare uno o più stream nel tipo di commit. - Per ogni stream, chiama
appendin un loop per scrivere batch di record. - Chiama
finalizeper ogni stream per rilasciarlo. Dopo aver chiamato questo metodo, non puoi scrivere altre righe nello stream. Questo passaggio è facoltativo nel tipo impegnato, ma aiuta a evitare di superare il limite di stream attivi. Per saperne di più, consulta Limita la frequenza di creazione degli stream.
Non puoi eliminare uno stream in modo esplicito. Gli stream seguono la durata (TTL) definita dal sistema:
- Uno stream impegnato ha un TTL di tre giorni se non è presente traffico sullo stream.
- Per impostazione predefinita, uno stream con buffer ha un TTL di sette giorni se non è presente traffico sullo stream.
Il seguente codice mostra come utilizzare il tipo committed:
Java
Per scoprire come installare e utilizzare la libreria client per BigQuery, consulta Librerie client di BigQuery. 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.
Node.js
Per scoprire come installare e utilizzare la libreria client per BigQuery, consulta Librerie client di BigQuery.
Per eseguire l'autenticazione in BigQuery, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura l'autenticazione per le librerie client.
Rust
Utilizza il formato Apache Arrow per importare i dati
Il seguente codice mostra come importare i dati utilizzando il formato Apache Arrow.
Python
Questo esempio mostra come importare una tabella PyArrow serializzata utilizzando lo stream predefinito. Per un esempio end-to-end più dettagliato, consulta l'esempio di PyArrow su GitHub.
from google.cloud.bigquery_storage_v1 import types as gapic_types
from google.cloud.bigquery_storage_v1.writer import AppendRowsStream
from google.cloud import bigquery_storage_v1
def append_rows_with_pyarrow(
pyarrow_table: pyarrow.Table,
project_id: str,
dataset_id: str,
table_id: str,
):
bqstorage_write_client = bigquery_storage_v1.BigQueryWriteClient()
# Create request_template.
request_template = gapic_types.AppendRowsRequest()
request_template.write_stream = (
f"projects/{project_id}/datasets/{dataset_id}/tables/{table_id}/_default"
)
arrow_data = gapic_types.AppendRowsRequest.ArrowData()
arrow_data.writer_schema.serialized_schema = (
pyarrow_table.schema.serialize().to_pybytes()
)
request_template.arrow_rows = arrow_data
# Create AppendRowsStream.
append_rows_stream = AppendRowsStream(
bqstorage_write_client,
request_template,
)
# Create request with table data.
request = gapic_types.AppendRowsRequest()
request.arrow_rows.rows.serialized_record_batch = (
pyarrow_table.to_batches()[0].serialize().to_pybytes()
)
# Send request.
future = append_rows_stream.send(request)
# Wait for result.
future.result()
Java
Per scoprire come installare e utilizzare la libreria client per BigQuery, consulta Librerie client di BigQuery. 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.
Rust