Cet exemple montre comment utiliser un tampon de protocole pour écrire des données dans une table BigQuery.
Exemple de code
Node.js
Avant d'essayer cet exemple, suivez les instructions de configuration pour Node.js du guide de démarrage rapide de BigQuery : Utiliser les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API BigQuery pour Node.js.
const {BigQueryWriteClient} = require('@google-cloud/bigquery-storage').v1;
const sample_data_pb = require('./sample_data_pb.js');
const protos = require('@google-cloud/bigquery-storage').protos.google.cloud
.bigquery.storage.v1;
const type = require('@google-cloud/bigquery-storage').protos.google.protobuf
.FieldDescriptorProto.Type;
const mode = require('@google-cloud/bigquery-storage').protos.google.cloud
.bigquery.storage.v1.WriteStream.Type;
async function appendRowsProto2() {
/**
* If you make updates to the sample_data.proto protocol buffers definition,
* run:
* protoc --js_out=import_style=commonjs,binary:. sample_data.proto
* from the /samples directory to generate the sample_data_pb module.
*/
const writeClient = new BigQueryWriteClient();
// So that BigQuery knows how to parse the serialized_rows, create a
// protocol buffer representation of your message descriptor.
const protoDescriptor = {};
protoDescriptor.name = 'SampleData';
protoDescriptor.field = [
{
name: 'bool_col',
number: 1,
type: type.TYPE_BOOL,
},
{
name: 'bytes_col',
number: 2,
type: type.TYPE_BYTES,
},
{
name: 'float64_col',
number: 3,
type: type.TYPE_FLOAT,
},
{
name: 'int64_col',
number: 4,
type: type.TYPE_INT64,
},
{
name: 'string_col',
number: 5,
type: type.TYPE_STRING,
},
{
name: 'date_col',
number: 6,
type: type.TYPE_INT32,
},
{
name: 'datetime_col',
number: 7,
type: type.TYPE_STRING,
},
{
name: 'geography_col',
number: 8,
type: type.TYPE_STRING,
},
{
name: 'numeric_col',
number: 9,
type: type.TYPE_STRING,
},
{
name: 'bignumeric_col',
number: 10,
type: type.TYPE_STRING,
},
{
name: 'time_col',
number: 11,
type: type.TYPE_STRING,
},
{
name: 'timestamp_col',
number: 12,
type: type.TYPE_INT64,
},
{
name: 'int64_list',
number: 13,
type: type.TYPE_INT64,
label: protos.TableFieldSchema.Mode.REPEATED,
},
{
name: 'struct_col',
number: 14,
typeName: 'SampleStruct',
type: type.TYPE_MESSAGE,
},
{
name: 'struct_list',
number: 15,
typeName: 'SampleStruct',
type: type.TYPE_MESSAGE,
label: protos.TableFieldSchema.Mode.REPEATED,
},
{
name: 'row_num',
number: 16,
type: type.TYPE_INT64,
label: protos.TableFieldSchema.Mode.REQUIRED,
},
];
protoDescriptor.nestedType = [
{
name: 'SampleStruct',
field: [
{
name: 'sub_int_col',
number: 1,
type: type.TYPE_INT64,
},
],
},
];
/**
* TODO(developer): Uncomment the following lines before running the sample.
*/
// projectId = 'my_project';
// datasetId = 'my_dataset';
// tableId = 'my_table';
const parent = `projects/${projectId}/datasets/${datasetId}/tables/${tableId}`;
try {
// Create a write stream to the given table.
let writeStream = {type: mode.PENDING};
let request = {
parent,
writeStream,
};
let [response] = await writeClient.createWriteStream(request);
console.log(`Stream created: ${response.name}`);
writeStream = response.name;
// This header is required so that the BigQuery Storage API
// knows which region to route the request to.
const options = {};
options.otherArgs = {};
options.otherArgs.headers = {};
options.otherArgs.headers[
'x-goog-request-params'
] = `write_stream=${writeStream}`;
// Append data to the given stream.
const stream = await writeClient.appendRows(options);
const responses = [];
stream.on('data', response => {
// Check for errors.
if (response.error) {
throw new Error(response.error.message);
}
console.log(response);
responses.push(response);
// Close the stream when all responses have been received.
if (responses.length === 3) {
stream.end();
}
});
stream.on('error', err => {
throw err;
});
stream.on('end', async () => {
// API call completed.
try {
[response] = await writeClient.finalizeWriteStream({
name: writeStream,
});
console.log(`Row count: ${response.rowCount}`);
[response] = await writeClient.batchCommitWriteStreams({
parent,
writeStreams: [writeStream],
});
console.log(response);
} catch (err) {
console.log(err);
}
});
let serializedRows = [];
// Row 1
let row = new sample_data_pb.SampleData();
row.setRowNum(1);
row.setBoolCol(true);
row.setBytesCol(Buffer.from('hello world'));
row.setFloat64Col(parseFloat('+123.45'));
row.setInt64Col(123);
row.setStringCol('omfg!');
serializedRows.push(row.serializeBinary());
// Row 2
row = new sample_data_pb.SampleData();
row.setRowNum(2);
row.setBoolCol(false);
serializedRows.push(row.serializeBinary());
// Row 3
row = new sample_data_pb.SampleData();
row.setRowNum(3);
row.setBytesCol(Buffer.from('later, gator'));
serializedRows.push(row.serializeBinary());
// Row 4
row = new sample_data_pb.SampleData();
row.setRowNum(4);
row.setFloat64Col(987.654);
serializedRows.push(row.serializeBinary());
// Row 5
row = new sample_data_pb.SampleData();
row.setRowNum(5);
row.setInt64Col(321);
serializedRows.push(row.serializeBinary());
// Row 6
row = new sample_data_pb.SampleData();
row.setRowNum(6);
row.setStringCol('octavia');
serializedRows.push(row.serializeBinary());
let protoRows = {
writerSchema: {protoDescriptor},
rows: {serializedRows},
};
// Set an offset to allow resuming this stream if the connection breaks.
// Keep track of which requests the server has acknowledged and resume the
// stream at the first non-acknowledged message. If the server has already
// processed a message with that offset, it will return an ALREADY_EXISTS
// error, which can be safely ignored.
// The first request must always have an offset of 0.
let offsetValue = 0;
// Construct request.
request = {
writeStream,
protoRows,
offset: {value: offsetValue},
};
// Send batch.
stream.write(request);
// Reset rows.
serializedRows = [];
// Row 7
row = new sample_data_pb.SampleData();
row.setRowNum(7);
row.setDateCol(1132896);
serializedRows.push(row.serializeBinary());
// Row 8
row = new sample_data_pb.SampleData();
row.setRowNum(8);
row.setDatetimeCol('2019-02-17 11:24:00.000');
serializedRows.push(row.serializeBinary());
// Row 9
row = new sample_data_pb.SampleData();
row.setRowNum(9);
row.setGeographyCol('POINT(5 5)');
serializedRows.push(row.serializeBinary());
// Row 10
row = new sample_data_pb.SampleData();
row.setRowNum(10);
row.setNumericCol('123456');
row.setBignumericCol('99999999999999999999999999999.999999999');
serializedRows.push(row.serializeBinary());
// Row 11
row = new sample_data_pb.SampleData();
row.setRowNum(11);
row.setTimeCol('18:00:00');
serializedRows.push(row.serializeBinary());
// Row 12
row = new sample_data_pb.SampleData();
row.setRowNum(12);
const timestamp = 1641700186564;
row.setTimestampCol(timestamp);
serializedRows.push(row.serializeBinary());
// Since this is the second request, you only need to include the row data.
// The name of the stream and protocol buffers DESCRIPTOR is only needed in
// the first request.
protoRows = {
rows: {serializedRows},
};
// Offset must equal the number of rows that were previously sent.
offsetValue = 6;
request = {
protoRows,
offset: {value: offsetValue},
};
// Send batch.
stream.write(request);
serializedRows = [];
// Row 13
row = new sample_data_pb.SampleData();
row.setRowNum(13);
row.addInt64List(1999);
row.addInt64List(2001);
serializedRows.push(row.serializeBinary());
// Row 14
row = new sample_data_pb.SampleData();
let sampleStruct = new sample_data_pb.SampleData.SampleStruct();
sampleStruct.setSubIntCol(99);
row.setRowNum(14);
row.setStructCol(sampleStruct);
serializedRows.push(row.serializeBinary());
// Row 15
row = new sample_data_pb.SampleData();
sampleStruct = new sample_data_pb.SampleData.SampleStruct();
sampleStruct.setSubIntCol(100);
const sampleStruct2 = new sample_data_pb.SampleData.SampleStruct();
sampleStruct2.setSubIntCol(101);
row.setRowNum(15);
row.addStructList(sampleStruct);
row.addStructList(sampleStruct2);
serializedRows.push(row.serializeBinary());
protoRows = {
rows: {serializedRows},
};
offsetValue = 12;
request = {
protoRows,
offset: {value: offsetValue},
};
// Send batch.
stream.write(request);
} catch (err) {
console.log(err);
}
}Python
Avant d'essayer cet exemple, suivez les instructions de configuration pour Python du guide de démarrage rapide de BigQuery : Utiliser les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API BigQuery pour Python.
"""
This code sample demonstrates using the low-level generated client for Python.
"""
import datetime
import decimal
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
# If you make updates to the sample_data.proto protocol buffers definition,
# run:
#
# protoc --python_out=. sample_data.proto
#
# from the samples/snippets directory to generate the sample_data_pb2 module.
from . import sample_data_pb2
def append_rows_proto2(project_id: str, dataset_id: str, table_id: str):
"""Create a write stream, write some sample data, and commit the stream."""
write_client = bigquery_storage_v1.BigQueryWriteClient()
parent = write_client.table_path(project_id, dataset_id, table_id)
write_stream = types.WriteStream()
# When creating the stream, choose the type. Use the PENDING type to wait
# until the stream is committed before it is visible. See:
# https://cloud.google.com/bigquery/docs/reference/storage/rpc/google.cloud.bigquery.storage.v1#google.cloud.bigquery.storage.v1.WriteStream.Type
write_stream.type_ = types.WriteStream.Type.PENDING
write_stream = write_client.create_write_stream(
parent=parent, write_stream=write_stream
)
stream_name = write_stream.name
# Create a template with fields needed for the first request.
request_template = types.AppendRowsRequest()
# The initial request must contain the stream name.
request_template.write_stream = stream_name
# So that BigQuery knows how to parse the serialized_rows, generate a
# protocol buffer representation of your 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
# Some stream types support an unbounded number of requests. Construct an
# AppendRowsStream to send an arbitrary number of requests to a stream.
append_rows_stream = writer.AppendRowsStream(write_client, request_template)
# Create a batch of row data by appending proto2 serialized bytes to the
# serialized_rows repeated field.
proto_rows = types.ProtoRows()
row = sample_data_pb2.SampleData()
row.row_num = 1
row.bool_col = True
row.bytes_col = b"Hello, World!"
row.float64_col = float("+inf")
row.int64_col = 123
row.string_col = "Howdy!"
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 2
row.bool_col = False
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 3
row.bytes_col = b"See you later!"
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 4
row.float64_col = 1000000.125
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 5
row.int64_col = 67000
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 6
row.string_col = "Auf Wiedersehen!"
proto_rows.serialized_rows.append(row.SerializeToString())
# Set an offset to allow resuming this stream if the connection breaks.
# Keep track of which requests the server has acknowledged and resume the
# stream at the first non-acknowledged message. If the server has already
# processed a message with that offset, it will return an ALREADY_EXISTS
# error, which can be safely ignored.
#
# The first request must always have an offset of 0.
request = types.AppendRowsRequest()
request.offset = 0
proto_data = types.AppendRowsRequest.ProtoData()
proto_data.rows = proto_rows
request.proto_rows = proto_data
response_future_1 = append_rows_stream.send(request)
# Create a batch of rows containing scalar values that don't directly
# correspond to a protocol buffers scalar type. See the documentation for
# the expected data formats:
# https://cloud.google.com/bigquery/docs/write-api#data_type_conversions
proto_rows = types.ProtoRows()
row = sample_data_pb2.SampleData()
row.row_num = 7
date_value = datetime.date(2021, 8, 12)
epoch_value = datetime.date(1970, 1, 1)
delta = date_value - epoch_value
row.date_col = delta.days
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 8
datetime_value = datetime.datetime(2021, 8, 12, 9, 46, 23, 987456)
row.datetime_col = datetime_value.strftime("%Y-%m-%d %H:%M:%S.%f")
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 9
row.geography_col = "POINT(-122.347222 47.651111)"
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 10
numeric_value = decimal.Decimal("1.23456789101112e+6")
row.numeric_col = str(numeric_value)
bignumeric_value = decimal.Decimal("-1.234567891011121314151617181920e+16")
row.bignumeric_col = str(bignumeric_value)
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 11
time_value = datetime.time(11, 7, 48, 123456)
row.time_col = time_value.strftime("%H:%M:%S.%f")
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 12
timestamp_value = datetime.datetime(
2021, 8, 12, 16, 11, 22, 987654, tzinfo=datetime.timezone.utc
)
epoch_value = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
delta = timestamp_value - epoch_value
row.timestamp_col = int(delta.total_seconds()) * 1000000 + int(delta.microseconds)
proto_rows.serialized_rows.append(row.SerializeToString())
# Since this is the second request, you only need to include the row data.
# The name of the stream and protocol buffers DESCRIPTOR is only needed in
# the first request.
request = types.AppendRowsRequest()
proto_data = types.AppendRowsRequest.ProtoData()
proto_data.rows = proto_rows
request.proto_rows = proto_data
# Offset must equal the number of rows that were previously sent.
request.offset = 6
response_future_2 = append_rows_stream.send(request)
# Create a batch of rows with STRUCT and ARRAY BigQuery data types. In
# protocol buffers, these correspond to nested messages and repeated
# fields, respectively.
proto_rows = types.ProtoRows()
row = sample_data_pb2.SampleData()
row.row_num = 13
row.int64_list.append(1)
row.int64_list.append(2)
row.int64_list.append(3)
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 14
row.struct_col.sub_int_col = 7
proto_rows.serialized_rows.append(row.SerializeToString())
row = sample_data_pb2.SampleData()
row.row_num = 15
sub_message = sample_data_pb2.SampleData.SampleStruct()
sub_message.sub_int_col = -1
row.struct_list.append(sub_message)
sub_message = sample_data_pb2.SampleData.SampleStruct()
sub_message.sub_int_col = -2
row.struct_list.append(sub_message)
sub_message = sample_data_pb2.SampleData.SampleStruct()
sub_message.sub_int_col = -3
row.struct_list.append(sub_message)
proto_rows.serialized_rows.append(row.SerializeToString())
request = types.AppendRowsRequest()
request.offset = 12
proto_data = types.AppendRowsRequest.ProtoData()
proto_data.rows = proto_rows
request.proto_rows = proto_data
# For each request sent, a message is expected in the responses iterable.
# This sample sends 3 requests, therefore expect exactly 3 responses.
response_future_3 = append_rows_stream.send(request)
# All three requests are in-flight, wait for them to finish being processed
# before finalizing the stream.
print(response_future_1.result())
print(response_future_2.result())
print(response_future_3.result())
# Shutdown background threads and close the streaming connection.
append_rows_stream.close()
# A PENDING type stream must be "finalized" before being committed. No new
# records can be written to the stream after this method has been called.
write_client.finalize_write_stream(name=write_stream.name)
# Commit the stream you created earlier.
batch_commit_write_streams_request = types.BatchCommitWriteStreamsRequest()
batch_commit_write_streams_request.parent = parent
batch_commit_write_streams_request.write_streams = [write_stream.name]
write_client.batch_commit_write_streams(batch_commit_write_streams_request)
print(f"Writes to stream: '{write_stream.name}' have been committed.")
Étape suivante
Pour rechercher et filtrer des exemples de code pour d'autres produits Google Cloud, consultez l'explorateur d'exemples Google Cloud.