linkedin/camus — pull request #10 by bcotton (2013-03-02 15:27:53+00:00)
Title: Support Arbitrary Partitions for the commit phase of camus

I want to get this out and get some feedback. I know there are plans to get rid of the EtlKey class, but this patch works with what I have now.

The use case is we have data embedded in records, beyond the timestamp, that we would like to use to partition the data by. Additionally we feed our Avro data directly to Hive, and need to use Hive-style partitioning (key=value/key2=value/..)

Configuration
Set the property etl.partitioner.class to your partitioner class. All topics will default to this class.
Set the property etl.partitioner.class. to override on a per-topic basis

The flow of data:

CamusWrapper was extended to contain a MapWritable to collect key/value pairs from the Record to be used in the partitioning path.

The MapWritable is taken from the CamusWriter and given to the EtlKey. This allows these values to show up in the EtlMultiOutputFormat.

This was done because in the places where the partitioner is called, the Record is not easly available.

I've introduced an interface, Partitioner, with the following signature:

String encodePartition(JobContext context, IEtlKey etlKey);
String generatePartitionedPath(JobContext context, String topic, int brokerId, int partitionId, String encodedPartition);

EtlMultiOutputFormat calls the first method on the partitioner when creating the "working" file to stream data from Kafka. The partitioner is handed the EtlKey for the record, and will encode the key/values pairs from the CamusWrapper into a string that will be embedded in the filename of the working file.

At commit time, the EtlMultiOutputCommitter will call the second method, generatePartitionedPath passing in some metadata and the encoded partition information. The parameter "encodedPartition" will be the same value that was returned from the first call to encodePartition().

One simple partitioner strategy would to "url encode" the key-values pairs into a string, and later simply substitute '&' with '/'.

I've extracted the current partitioning behavior into a DefaultPartitioner class, which will be the default behavior if not configured. All other metadata encoded into the original code (counts offsets etc) remain and are not part of the partitioner's purview.

Implementation Notes:

It seemed natural the place the Partitioner interface into the camus-api module, I then needed to extract an interface for EtlKey, IEtlKey, so instances of Partitioner would have access to that data. This can change to expose just the primitive types, but may make the interface cluttered.