Google BigQuery est l'entrepôt de données serverless de Google, conçu pour exécuter des requêtes analytiques sur de très grandes tables. Lorsque vous voulez extraire des données de table de BigQuery vers votre propre application, le chemin le plus rapide est la BigQuery Storage Read API, un service gRPC qui diffuse les lignes directement depuis le stockage au lieu de les paginer via un endpoint REST. sgcWebSockets Enterprise fournit un client gRPC BigQuery typé qui repose sur TsgcGRPCClient, ce qui vous permet de lire une table BigQuery depuis Delphi et C++Builder sans aucun runtime gRPC externe.
Comment ça fonctionne
La Storage Read API est un service gRPC nommé google.cloud.bigquery.storage.v1.BigQueryRead. gRPC, ce sont des Protocol Buffers encadrés sur HTTP/2, donc le client fonctionne sur la pile HTTP/2 de sgcWebSockets. Vous pointez un TsgcHTTP2Client vers bigquerystorage.googleapis.com sur le port 443 avec TLS, l'affectez à un TsgcGRPCClient, et le client gRPC gère l'encadrement des messages, les en-têtes, les délais d'attente et les trailers à votre place.
Lire une table est un schéma à deux appels. D'abord vous appelez CreateReadSession, un appel unaire qui demande à BigQuery d'ouvrir une session sur une table et de renvoyer un ou plusieurs flux de lecture. Ensuite vous appelez ReadRows sur un flux, un appel à diffusion serveur : vous envoyez une requête et le serveur renvoie une séquence de lots de lignes jusqu'à la fin du flux. Les messages BigQuery typés dans sgcGRPC_Google_BigQuery construisent les protobufs de requête et analysent les réponses, donc vous travaillez avec des classes Delphi au lieu d'octets assemblés à la main.
Authentification
BigQuery est authentifié avec un compte de service Google Cloud. La démo charge le JSON du compte de service, construit un JWT auto-signé et l'échange contre un jeton d'accès, puis envoie ce jeton en tant que métadonnées gRPC sur chaque appel. Comme un JWT auto-signé de compte de service est lié à une audience, le jeton doit cibler l'endpoint BigQuery Storage afin qu'il soit accepté par bigquerystorage.googleapis.com.
GoogleCloud.GoogleCloudOptions.Authentication := gcaJWT;
GoogleCloud.GoogleCloudOptions.JWT.ClientEmail := ClientEmail;
GoogleCloud.GoogleCloudOptions.JWT.PrivateKeyId := PrivateKeyId;
GoogleCloud.GoogleCloudOptions.JWT.PrivateKey.Text := PrivateKey;
GoogleCloud.GoogleCloudOptions.JWT.ProjectId := ProjectId;
GoogleCloud.GoogleCloudOptions.JWT.API_Endpoint :=
'https://bigquerystorage.googleapis.com/';
// once the token is acquired, attach it to every gRPC call
GRPC.DefaultMetadata.Add('authorization', 'Bearer ' + Token);Configurer le client
Un canal gRPC est une connexion HTTP/2. Créez le transport, pointez-le vers l'hôte BigQuery Storage, puis créez le client gRPC et affectez le transport à sa propriété Client. Le type de contenu est le format de fil proto binaire et la compression est laissée désactivée.
uses
sgcHTTP2_Client, sgcGRPC_Client, sgcGRPC_Types, sgcGRPC_Google_BigQuery;
var
HTTP2: TsgcHTTP2Client;
GRPC: TsgcGRPCClient;
begin
HTTP2 := TsgcHTTP2Client.Create(nil);
HTTP2.Host := 'bigquerystorage.googleapis.com';
HTTP2.Port := 443;
HTTP2.TLS := True;
GRPC := TsgcGRPCClient.Create(nil);
GRPC.Client := HTTP2;
GRPC.ChannelOptions.ContentType := grpcProto;
GRPC.ChannelOptions.Compression := grpcNoCompression;
HTTP2.Active := True;
end;Créer une session de lecture
Pour ouvrir une session, vous renseignez un TsgcGRPCBigQueryCreateReadSessionRequest. Le Parent est votre projet de facturation, le ReadSession.Table est le chemin de table pleinement qualifié, et DataFormat sélectionne le format de fil des lignes (1 = Avro, 2 = Arrow). MaxStreamCount plafonne le nombre de flux de lecture parallèles que BigQuery renvoie. Sérialisez la requête avec ToBytes et envoyez-la en tant que Call unaire ; la réponse est une ReadSession que vous chargez avec LoadFromBytes pour relire les noms de flux.
var
oRequest: TsgcGRPCBigQueryCreateReadSessionRequest;
oResponse: TsgcGRPCResponse;
oSession: TsgcGRPCBigQueryReadSession;
begin
oRequest := TsgcGRPCBigQueryCreateReadSessionRequest.Create;
try
oRequest.Parent := 'projects/' + ProjectId;
oRequest.ReadSession.Table := 'projects/' + ProjectId +
'/datasets/' + Dataset + '/tables/' + Table;
oRequest.ReadSession.DataFormat := 1; { AVRO }
oRequest.MaxStreamCount := 1;
oResponse := GRPC.Call(
'google.cloud.bigquery.storage.v1.BigQueryRead',
'CreateReadSession', oRequest.ToBytes);
if oResponse.StatusCode = grpcOK then
begin
oSession := TsgcGRPCBigQueryReadSession.Create;
try
oSession.LoadFromBytes(oResponse.Data);
// keep the first read stream name for ReadRows
if oSession.StreamCount > 0 then
FReadStreamName := oSession.Stream(0).Name;
finally
oSession.Free;
end;
end;
finally
oRequest.Free;
end;
end;Diffuser les lignes
Avec un nom de flux en main, vous appelez ReadRows. Définissez ReadStream sur le nom de flux et un Offset de départ optionnel, puis envoyez-le avec CallAsync pour une réponse à diffusion serveur. Chaque lot que le serveur renvoie déclenche OnGRPCStreamMessage, et OnGRPCStreamEnd se déclenche une fois lorsque le flux est terminé.
var
oRequest: TsgcGRPCBigQueryReadRowsRequest;
begin
oRequest := TsgcGRPCBigQueryReadRowsRequest.Create;
try
oRequest.ReadStream := FReadStreamName;
oRequest.Offset := 0;
GRPC.CallAsync('google.cloud.bigquery.storage.v1.BigQueryRead',
'ReadRows', oRequest.ToBytes);
finally
oRequest.Free;
end;
end;
procedure TForm1.OnGRPCStreamMessage(Sender: TObject;
const aMessage: TsgcGRPCStreamMessage; var aCancel: Boolean);
var
oResponse: TsgcGRPCBigQueryReadRowsResponse;
begin
oResponse := TsgcGRPCBigQueryReadRowsResponse.Create;
try
oResponse.LoadFromBytes(aMessage.Data);
if oResponse.AvroRows.RowCount > 0 then
// oResponse.AvroRows.SerializedBinaryRows holds the Avro block
else if oResponse.ArrowRecordBatch.RowCount > 0 then
// oResponse.ArrowRecordBatch.SerializedRecordBatch holds the Arrow batch
finally
oResponse.Free;
end;
end;Formats de lignes Avro et Arrow
BigQuery ne renvoie pas les lignes sous forme de champs analysés sur le fil. Il renvoie des blocs sérialisés dans l'un de deux formats en colonnes, et vous choisissez le format lorsque vous créez la session. Avec DataFormat := 1 la réponse porte AvroRows, dont le champ SerializedBinaryRows contient un bloc encodé en Avro et RowCount vous indique combien de lignes il contient. Avec DataFormat := 2 la réponse porte un ArrowRecordBatch, dont le champ SerializedRecordBatch contient un record batch Apache Arrow. Vous décodez le bloc avec la bibliothèque Avro ou Arrow de votre choix. Le schéma de session, renvoyé avec les flux, décrit les noms et types de colonnes afin que vous sachiez ce que chaque bloc contient.
Filtrer et projeter
Vous n'êtes pas obligé de lire des lignes entières. La session de lecture porte un objet ReadOptions qui vous permet de pousser une projection et un filtre jusqu'à BigQuery afin qu'il envoie moins de données. Ajoutez des noms de colonnes à SelectedFields pour ne projeter que les colonnes dont vous avez besoin, et définissez RowRestriction sur un prédicat de style SQL pour filtrer les lignes côté serveur avant qu'elles ne soient diffusées.
oRequest.ReadSession.ReadOptions.SelectedFields.Add('name');
oRequest.ReadSession.ReadOptions.SelectedFields.Add('state');
oRequest.ReadSession.ReadOptions.RowRestriction := 'state = "CA"';Disponibilité
Le client gRPC BigQuery fait partie de l'édition Enterprise de sgcWebSockets et fonctionne sous Windows, macOS, Linux, iOS et Android. Un exemple prêt à l'emploi, avec le chargeur de compte de service, le flux create-session et read-rows montré ci-dessus, se trouve dans Demos\21.GRPC\16.BigQuery. La référence complète, y compris les autres clients gRPC Google Cloud, est sur la page produit du client gRPC.
Des questions ou des commentaires ? Contactez-nous. Vous recevrez une réponse des personnes qui ont écrit le code.