A PTransform for writing a PCollection to file-based sink, where files represent "buckets" of elements deterministically assigned by BucketMetadata based on a key extraction function. The elements in each bucket are written in sorted order according to the same key.

This transform is intended to be used in conjunction with the SortedBucketSource transform. Any two datasets written with SortedBucketSink using the same bucketing scheme can be joined by simply sequentially reading and merging files, thus eliminating the shuffle required by GroupByKey-based transforms. This is ideal for datasets that will be written once and read many times with a predictable join key, i.e. user event data.

Transform steps

SortedBucketSink maps over each element, extracts a byte[] representation of its sorting key using BucketMetadata#extractKeyPrimary(Object), and assigns it to an Integer bucket using BucketMetadata#getBucketId(byte[]). Next, a GroupByKey transform is applied to create a PCollection of N elements, where N is the number of buckets specified by BucketMetadata#getNumBuckets(), then a SortBucketShard transform is used to sort elements within each bucket group, optionally sorting by the secondary key bytes from BucketMetadata#getKeyBytesSecondary(Object). Finally, the write operation is performed, where each bucket is first written to a SortedBucketSink#tempDirectory and then copied to its final destination.

A JSON-serialized form of BucketMetadata is also written, which is required in order to join SortedBucketSinks using the SortedBucketSource transform.

Bucketing properties and hot keys

Bucketing properties are specified in BucketMetadata. The number of buckets, N, must be a power of two and should be chosen such that each bucket can fit in a worker node's memory. Note that the SortValues transform will try to sort in-memory and fall back to an ExternalSorter if needed.

Each bucket can be further sharded to reduce the impact of hot keys, by specifying BucketMetadata#getNumShards().