Hadoop Pre-processing Job

### Smart PBNJ
with contributions from @jackjlli (prototype: https://github.com/apache/incubator-pinot/pull/4253)

**Motivation**
Currently, users have to do a lot of pre-processing and have to know a lot about the internals of Pinot in order to generate the Pinot segment format. We want to prevent all of this and hope this project can improve cluster stability & reduce/eliminate issues with many small segments.

The scope of this project includes:
- Sorting and partitioning for our users
- Generating segments of a proper size for users. This prevents problems with having many small segments or super large segments.

### Design

**Configurations**
If set, we will fetch the following parameters from a new section of table config called hadoopConfigs.
enable.partitioning
enable.sorting
min.num.output.files
max.num.records

_enable.partitioning_
False by default. If true, partition column, partition function and number of partitions within table config will be used to generate segments that contain data from a single partition. Number of partitions determines number of reducers.
_enable.sorting_
If this parameter is set and a valid sorted column is declared within the table indexing config (sometimes, people declare multiple which is invalid), we will sort user data.
_min.num.output.files_
This parameter determines the number of reducers when partitioning is disabled.
_max.num.records_
This parameter refers to the maximum records that can be in a file. When not set, each reducer will generate one output file. When set, each reducer will start another file once it meets the configured limit. Caveat - see Smart Segment Resizing

**Smart Segment Resizing**
We will use max.num.records to resize segments to create performant datasets. This parameter will help us avoid problems with too many poorly-sized Pinot segments that will cause us to use many server threads.

One problem with this approach will be the possibility of segments that are unbalanced in size. We will correct this by setting a minimum percentage threshold for which a small number of extra records will be folded into the latest file created within that reducer.

**Multiple Formats**
To support each output format, we need to be plug in different format-specific classes within the pre-processing job.
job.setMapOutputKeyClass(AvroKey.class)
job.setMapOutputValueClass(AvroValue.class)
job.setOutputKeyClass(AvroKey.class)
AvroMultipleOutputs….
For ORC, for example, we have these classes available.
http://hadoopathome.logdown.com/posts/277986-using-multipleoutputs-with-orc-in-mapreduce
https://orc.apache.org/docs/mapreduce.html

Depending on the format the client configures in the configs, we will pick the correct classes to set.

We will be able to add and support any format that supports these Hadoop mapreduce classes.

**Backfill/Naming**

Because we are mutating user segments, we need to ensure that
Segments that span different time partitions are not resized into the same segment, eg: multiple days
Extra partitions are deleted when users re-push the same time partition / when users refresh their segments.

To prevent 1), I will check the minimum and maximum timestamps of each file and ensure that they fit into the same time partition.

To prevent 2), We will query the controller to determine the segments present before pushing. After pushing, we will query again and delete extra segments present.
There is a possible race condition, but I do not think it will be a problem because:
Case 1: Segments are being pushed elsewhere - There is a current contract with clients that this does not happen - if this happened, segment push itself may end up with inconsistent data (some from each push), and we can only solve this once we can figure out which "batch" segments are coming from.
Case 2: someone else is deleting segments - Worst case, we will have **FEWER** segments, not more segments.

**Contraindications**
As mentioned above, users who want to partition their data will not be able to use segment resizing, as they are incompatible.

**Appendix**
Reducer Output Filename Format
filePrefix fileOffset -x-yyyyy
File prefix
A random string with five characters that is common across all files generated by a reducer.
File offset
An offset to denote the index of the output file within the reducer. If “max.num.records” is not set, file offset is empty. If “max.num.records” is correctly specified, the file offset will start from 0 to N-1, where N is the total number of output files from the same reducer.
X
r for reducer
yyyyy
Mapper/reducer task number