Note: Additional details and implementation specifics will be added over time.
A requirement/use-case is to continuously process partitions of a PartitionedFileSetDataset from within a periodically running MapReduce job, scheduled by a Workflow. This means that each time the Workflow runs, it should process all of the partitions created since the last one it processed, so that it has processed all existing partitions after running. In order to do this, we will be adding an index on creation time of the partitions, as they are added to the dataset, in order to avoid inefficiently scanning all of the partitions each run. Then, one possibility is that the workflow will be responsible for maintaining the end timestamp used when requesting the partitions by time (lastProcessedTimestamp). In the next run of the workflow, it can then simply request from the dataset the partitions that have been created after lastProcessedTimestamp up until the then current time.
However, there is an issue with this idea - a partition could have had a creation time smaller/earlier than lastProcessedTimestamp, but not have been committed/visible at the time the workflow requested partitions up until lastProcessedTimestamp. So, even though the partition became visible after lastProcessedTimestamp, the next scan from lastProcessedTimestamp onwards will not include this partition because the partition's creation time will not match the time range. While this approach is simple, it has the possibility to omit the processing of partitions, and so it is not good enough.
An improved solution for this would be for the workflow consuming the partitions to keep track of two things each time it consumes partitions: the set of in-progress transactions as well as the write pointer of the transaction in which the partitions are consumed. The write pointer will be used as a high water mark from which to start scanning for new partitions the next time partitions are polled. The set of in-progress transactions will indicate points at which new partitions are created, but before the previously mentioned high water mark.
Then, each time the partition consumer runs and polls for additional partitions, it will do a scan by time starting from the previous transaction's write pointe up until the current time. It will also have to check if any of the previously in-progress transactions are no longer running and if they correspond to new partitions, by doing lookups or a scan on the index table. This ensures that partitions which were added in transactions that committed later are still processed.
Note that these are just the APIs defined on the PartitionedFileSet class and a simple consumer.
APIs:
PartitionFileSet.java:
public PartitionConsumerResult consumePartitions(PartitionConsumerState partitionConsumerState);
PartitionConsumerResult.java:
Iterator<Partition> partitionsIterator; PartitionConsumerState partitionConsumerState;
PartitionConsumerState.java:
Long startVersion; long[] versionsToCheck;
PartitionConsumer.java:
public Iterator<Partition> consumePartitions();
This PartitionConsumer would be a simple consumer (without persistence persistence) which calls PartitionFileSet's consumerPartitions method with a PartitionConsumerState and keeps track of the previously returned PartitionConsumerState, to pass in the next call. Users of this PartitionConsumer (for instance, a periodically scheduled workflow or a thread in a scheduler) can manage persistence of it, with this approach.
Related JIRAs: