add partiton logic.

Author: civitaspo

Diff for: README.md

@@ -15,6 +15,7 @@ Read files on Hdfs.
 - **input_path** file path on Hdfs. you can use glob and Date format like `%Y%m%d/%s`.
 - **rewind_seconds** When you use Date format in input_path property, the format is executed by using the time which is Now minus this property.
 - **partition** when this is true, partition input files and increase task count. (default: `true`)
+- **num_partitions** number of partitions. (default: `Runtime.getRuntime().availableProcessors()`)
 
 ## Example
 
@@ -32,6 +33,7 @@ in:
   input_path: /user/embulk/test/%Y-%m-%d/*
   rewind_seconds: 86400
   partition: true
+  num_partitions: 30
   decoders:
     - {type: gzip}
   parser:
@@ -53,6 +55,8 @@ in:
 ```
 
 ## Note
+- The parameter **num_partitions** is the approximate value. The actual num_partitions is larger than this parameter.
+  - see: [The Partitioning Logic](#partition_logic)
 - the feature of the partition supports only 3 line terminators.
   - `\n`
   - `\r`
@@ -61,6 +65,36 @@ in:
 ## The Reference Implementation
 - [hito4t/embulk-input-filesplit](https://github.com/hito4t/embulk-input-filesplit)
 
+##<a id="partition_logic">The Partitioning Logic</a>
+
+```
+int partitionSizeByOneTask = totalFileLength / approximateNumPartitions;
+
+/*
+...
+*/
+
+    int numPartitions;
+    if (path.toString().endsWith(".gz") || path.toString().endsWith(".bz2") || path.toString().endsWith(".lzo")) {
+        // if the file is compressed, skip partitioning.
+        numPartitions = 1;
+    }
+    else if (!task.getPartition()) {
+        // if no partition mode, skip partitioning.
+        numPartitions = 1;
+    } 
+    else {
+        // equalize the file size per task as much as possible.
+        numPartitions = ((fileLength - 1) / partitionSizeByOneTask) + 1;
+    }
+
+/*
+...
+*/
+
+```
+
+
 ## Build
 
 ```