Read and append to objects in zonal buckets

You can read from and append data to objects in zonal buckets, which are made available through Rapid Bucket . Zonal buckets use the Rapid storage class , a high-performance storage class that's optimized for I/O-intensive workloads. Appendable objects let you incrementally add data to a file without needing to rewrite the entire object, helping you more effectively manage continuously flowing data.

This page describes how to read and make appends to objects stored in zonal buckets. It shows you how to perform the following operations:

Create and write to an appendable object.
Read appendable objects.
Pause, resume, and finalize appendable objects.
Read the tail end of appendable objects.

Before you use this page, you might want to read the following resources:

Learn how to create a zonal bucket .
Upload objects to a bucket .

Make appendable writes to objects

Client libraries

C++

For more information, see the Cloud Storage C++ API reference documentation .

To authenticate to Cloud Storage, set up Application Default Credentials. For more information, see Set up authentication for client libraries .

The following sample uploads an appendable object:

  namespace 
  
 gcs 
  
 = 
  
 google 
 :: 
 cloud 
 :: 
 storage 
 ; 
 auto 
  
 coro 
  
 = 
  
 []( 
 gcs 
 :: 
 AsyncClient 
&  
 client 
 , 
  
 std 
 :: 
 string 
  
 bucket_name 
 , 
  
 std 
 :: 
 string 
  
 object_name 
 ) 
  
 - 
>  
 google 
 :: 
 cloud 
 :: 
 future<google 
 :: 
 storage 
 :: 
 v2 
 :: 
 Object 
>  
 { 
  
 auto 
  
 [ 
 writer 
 , 
  
 token 
 ] 
  
 = 
  
 ( 
 co_await 
  
 client 
 . 
 StartAppendableObjectUpload 
 ( 
  
 gcs 
 :: 
 BucketName 
 ( 
 std 
 :: 
 move 
 ( 
 bucket_name 
 )), 
  
 std 
 :: 
 move 
 ( 
 object_name 
 ))) 
  
 . 
 value 
 (); 
  
 std 
 :: 
 cout 
 << 
 "Appendable upload started for object " 
 << 
 object_name 
 << 
 " 
 \n 
 " 
 ; 
  
 token 
  
 = 
  
 ( 
 co_await 
  
 writer 
 . 
 Write 
 ( 
 std 
 :: 
 move 
 ( 
 token 
 ), 
  
 gcs 
 :: 
 WritePayload 
 ( 
 "Some data 
 \n 
 " 
 ))) 
  
 . 
 value 
 (); 
  
 std 
 :: 
 cout 
 << 
 "Wrote initial data. 
 \n 
 " 
 ; 
  
 // Flush the buffered data to the service. This is not a terminal 
  
 // operation. The writer can be used after the flush completes. 
  
 // After a flush, the data is visible to readers. 
  
 auto 
  
 flush_status 
  
 = 
  
 co_await 
  
 writer 
 . 
 Flush 
 (); 
  
 if 
  
 ( 
 ! 
 flush_status 
 . 
 ok 
 ()) 
  
 throw 
  
 std 
 :: 
 runtime_error 
 ( 
 flush_status 
 . 
 message 
 ()); 
  
 std 
 :: 
 cout 
 << 
 "Flush completed. Persisted size is now " 
 << 
 absl 
 :: 
 get<std 
 :: 
 int64_t 
> ( 
 writer 
 . 
 PersistedState 
 ()) 
 << 
 " 
 \n 
 " 
 ; 
  
 // The writer is still open. We can write more data. 
  
 token 
  
 = 
  
 ( 
 co_await 
  
 writer 
 . 
 Write 
 ( 
 std 
 :: 
 move 
 ( 
 token 
 ), 
  
 gcs 
 :: 
 WritePayload 
 ( 
 "Some more data 
 \n 
 " 
 ))) 
  
 . 
 value 
 (); 
  
 std 
 :: 
 cout 
 << 
 "Wrote more data. 
 \n 
 " 
 ; 
  
 // Finalize the upload to make it a regular object. 
  
 co_return 
  
 ( 
 co_await 
  
 writer 
 . 
 Finalize 
 ( 
 std 
 :: 
 move 
 ( 
 token 
 ))). 
 value 
 (); 
 };

Go

For more information, see the Cloud Storage Go API reference documentation .

To authenticate to Cloud Storage, set up Application Default Credentials. For more information, see Set up authentication for client libraries .

The following sample uploads an appendable object:

  import 
  
 ( 
  
 "context" 
  
 "fmt" 
  
 "io" 
  
 "time" 
  
 "cloud.google.com/go/storage" 
  
 "cloud.google.com/go/storage/experimental" 
 ) 
 // createAndWriteAppendableObject creates and uploads a new appendable object in 
 // a rapid bucket. The object will not be finalized. 
 func 
  
 createAndWriteAppendableObject 
 ( 
 w 
  
 io 
 . 
  Writer 
 
 , 
  
 bucket 
 , 
  
 object 
  
 string 
 ) 
  
 error 
  
 { 
  
 // bucket := "bucket-name" 
  
 // object := "object-name" 
  
 ctx 
  
 := 
  
 context 
 . 
 Background 
 () 
  
 client 
 , 
  
 err 
  
 := 
  
 storage 
 . 
  NewGRPCClient 
 
 ( 
 ctx 
 , 
  
 experimental 
 . 
  WithZonalBucketAPIs 
 
 ()) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "storage.NewGRPCClient: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 defer 
  
 client 
 . 
 Close 
 () 
  
 ctx 
 , 
  
 cancel 
  
 := 
  
 context 
 . 
 WithTimeout 
 ( 
 ctx 
 , 
  
 time 
 . 
 Second 
 * 
 10 
 ) 
  
 defer 
  
 cancel 
 () 
  
 // Create a Writer and write some data. 
  
 writer 
  
 := 
  
 client 
 . 
  Bucket 
 
 ( 
 bucket 
 ). 
  Object 
 
 ( 
 object 
 ). 
  NewWriter 
 
 ( 
 ctx 
 ) 
  
 if 
  
 _ 
 , 
  
 err 
  
 := 
  
 writer 
 . 
  Write 
 
 ([] 
 byte 
 ( 
 "Some data\n" 
 )); 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "Writer.Write: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 // Flush the buffered data to the service. This is not a terminal 
  
 // operation. The Writer can be used after the flush completes. 
  
 // After a flush, the data is visible to readers. 
  
 size 
 , 
  
 err 
  
 := 
  
 writer 
 . 
  Flush 
 
 () 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "Writer.Flush: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Flush completed. Persisted size is now %d\n" 
 , 
  
 size 
 ) 
  
 // The Writer is still open. We can write more data. 
  
 if 
  
 _ 
 , 
  
 err 
  
 := 
  
 writer 
 . 
  Write 
 
 ([] 
 byte 
 ( 
 "Some more data\n" 
 )); 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "Writer.Write: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 // Close the Writer to flush any remaining buffered data. 
  
 // The object will be unfinalized, which means another writer can 
  
 // later append to the object. 
  
 if 
  
 err 
  
 := 
  
 writer 
 . 
 Close 
 (); 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "Writer.Close: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Uploaded object %v\n" 
 , 
  
 object 
 ) 
  
 return 
  
 nil 
 }

Java

For more information, see the Cloud Storage Java API reference documentation .

To authenticate to Cloud Storage, set up Application Default Credentials. For more information, see Set up authentication for client libraries .

The following sample uploads an appendable object:

  import 
  
 com.google.cloud.storage. BlobAppendableUpload 
 
 ; 
 import 
  
 com.google.cloud.storage. BlobAppendableUpload 
. AppendableUploadWriteableByteChannel 
 
 ; 
 import 
  
 com.google.cloud.storage. BlobAppendableUploadConfig 
 
 ; 
 import 
  
 com.google.cloud.storage. BlobAppendableUploadConfig 
. CloseAction 
 
 ; 
 import 
  
 com.google.cloud.storage. BlobId 
 
 ; 
 import 
  
 com.google.cloud.storage. BlobInfo 
 
 ; 
 import 
  
 com.google.cloud.storage. FlushPolicy 
 
 ; 
 import 
  
 com.google.cloud.storage. Storage 
 
 ; 
 import 
  
 com.google.cloud.storage. StorageOptions 
 
 ; 
 import 
  
 com.google.common.io.ByteStreams 
 ; 
 import 
  
 java.io.IOException 
 ; 
 import 
  
 java.nio.channels.FileChannel 
 ; 
 import 
  
 java.nio.channels.ReadableByteChannel 
 ; 
 import 
  
 java.nio.file.Paths 
 ; 
 import 
  
 java.util.Locale 
 ; 
 public 
  
 class 
 CreateAndWriteAppendableObject 
  
 { 
  
 public 
  
 static 
  
 void 
  
 createAndWriteAppendableObject 
 ( 
  
 String 
  
 bucketName 
 , 
  
 String 
  
 objectName 
 , 
  
 String 
  
 filePath 
 ) 
  
 throws 
  
 Exception 
  
 { 
  
 // The ID of your GCS bucket 
  
 // String bucketName = "your-unique-bucket-name"; 
  
 // The ID of your GCS object 
  
 // String objectName = "your-object-name"; 
  
 // The path to the file to upload 
  
 // String filePath = "path/to/your/file"; 
  
 try 
  
 ( 
  Storage 
 
  
 storage 
  
 = 
  
  StorageOptions 
 
 . 
  grpc 
 
 (). 
 build 
 (). 
  getService 
 
 ()) 
  
 { 
  
  BlobId 
 
  
 blobId 
  
 = 
  
  BlobId 
 
 . 
 of 
 ( 
 bucketName 
 , 
  
 objectName 
 ); 
  
  BlobInfo 
 
  
 blobInfo 
  
 = 
  
  BlobInfo 
 
 . 
 newBuilder 
 ( 
 blobId 
 ). 
 build 
 (); 
  
 int 
  
 flushSize 
  
 = 
  
 64 
  
 * 
  
 1000 
 ; 
  
  FlushPolicy 
 
 . 
  MaxFlushSizeFlushPolicy 
 
  
 flushPolicy 
  
 = 
  
  FlushPolicy 
 
 . 
  maxFlushSize 
 
 ( 
 flushSize 
 ); 
  
  BlobAppendableUploadConfig 
 
  
 config 
  
 = 
  
  BlobAppendableUploadConfig 
 
 . 
 of 
 () 
  
 . 
  withCloseAction 
 
 ( 
  CloseAction 
 
 . 
 FINALIZE_WHEN_CLOSING 
 ) 
  
 . 
  withFlushPolicy 
 
 ( 
 flushPolicy 
 ); 
  
  BlobAppendableUpload 
 
  
 uploadSession 
  
 = 
  
 storage 
 . 
  blobAppendableUpload 
 
 ( 
 blobInfo 
 , 
  
 config 
 ); 
  
 try 
  
 ( 
  AppendableUploadWriteableByteChannel 
 
  
 channel 
  
 = 
  
 uploadSession 
 . 
  open 
 
 (); 
  
 ReadableByteChannel 
  
 readableByteChannel 
  
 = 
  
 FileChannel 
 . 
 open 
 ( 
 Paths 
 . 
 get 
 ( 
 filePath 
 ))) 
  
 { 
  
 ByteStreams 
 . 
 copy 
 ( 
 readableByteChannel 
 , 
  
 channel 
 ); 
  
 // Since the channel is in a try-with-resources block, channel.close() 
  
 // will be implicitly called here, which triggers the finalization. 
  
 } 
  
 catch 
  
 ( 
 IOException 
  
 ex 
 ) 
  
 { 
  
 throw 
  
 new 
  
 IOException 
 ( 
 "Failed to upload to object " 
  
 + 
  
 blobId 
 . 
  toGsUtilUri 
 
 (), 
  
 ex 
 ); 
  
 } 
  
  BlobInfo 
 
  
 result 
  
 = 
  
 storage 
 . 
 get 
 ( 
 blobId 
 ); 
  
 System 
 . 
 out 
 . 
 printf 
 ( 
  
 Locale 
 . 
 US 
 , 
  
 "Object %s successfully uploaded" 
 , 
  
  result 
 
 . 
  getBlobId 
 
 (). 
  toGsUtilUriWithGeneration 
 
 ()); 
  
 } 
  
 } 
 }

Python

For more information, see the Cloud Storage Python API reference documentation .

To authenticate to Cloud Storage, set up Application Default Credentials. For more information, see Set up authentication for client libraries .

The following sample uploads an appendable object:

  async 
 def 
  
 storage_create_and_write_appendable_object 
 ( 
 bucket_name 
 , 
 object_name 
 , 
 grpc_client 
 = 
 None 
 ): 
  
 """Uploads an appendable object to zonal bucket. 
 grpc_client: an existing grpc_client to use, this is only for testing. 
 """ 
 if 
 grpc_client 
 is 
 None 
 : 
 grpc_client 
 = 
 AsyncGrpcClient 
 () 
 writer 
 = 
 AsyncAppendableObjectWriter 
 ( 
 client 
 = 
 grpc_client 
 , 
 bucket_name 
 = 
 bucket_name 
 , 
 object_name 
 = 
 object_name 
 , 
 generation 
 = 
 0 
 , 
 # throws `FailedPrecondition` if object already exists. 
 ) 
 # This creates a new appendable object of size 0 and opens it for appending. 
 await 
 writer 
 . 
 open 
 () 
 # appends data to the object 
 # you can perform `.append` multiple times as needed. Data will be appended 
 # to the end of the object. 
 await 
 writer 
 . 
 append 
 ( 
 b 
 "Some data" 
 ) 
 # Once all appends are done, close the gRPC bidirectional stream. 
 await 
 writer 
 . 
 close 
 () 
 print 
 ( 
 f 
 "Appended object 
 { 
 object_name 
 } 
 created of size 
 { 
 writer 
 . 
 persisted_size 
 } 
 bytes." 
 )

Read and append to objects in zonal buckets

Make appendable writes to objects

Client libraries

C++

Go

Java

Python

Read objects

Client libraries

C++

Go

Java

Python

Pause and resume an object

Client libraries

C++

Go

Java

Python

Finalize an object

Client libraries

C++

Go

Java

Python

Read the tail of an object

Client libraries

C++

Go

Java

Python

What's next

Read and append to objects in zonal buckets Stay organized with collections Save and categorize content based on your preferences.

Make appendable writes to objects

Client libraries

C++

Go

Java

Python

Read objects

Client libraries

C++

Go

Java

Python

Pause and resume an object

Client libraries

C++

Go

Java

Python

Finalize an object

Client libraries

C++

Go

Java

Python

Read the tail of an object

Client libraries

C++

Go

Java

Python

What's next

Read and append to objects in zonal buckets