'maxBytesInMemory' tuningConfig introduced for ingestion tasks

docs/content/development/extensions-core/kafka-ingestion.md

@@ -115,7 +115,8 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |Field|Type|Description|Required|
 |-----|----|-----------|--------|
 |`type`|String|The indexing task type, this should always be `kafka`.|yes|
-|`maxRowsInMemory`|Integer|The number of rows to aggregate before persisting. This number is the post-aggregation rows, so it is not equivalent to the number of input events, but the number of aggregated rows that those events result in. This is used to manage the required JVM heap size. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|no (default == 75000)|
+|`maxRowsInMemory`|Integer|The number of rows to aggregate before persisting. This number is the post-aggregation rows, so it is not equivalent to the number of input events, but the number of aggregated rows that those events result in. This is used to manage the required JVM heap size. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists). Normally user does not need to set this, but depending on the nature of data, if rows are short in terms of bytes, user may not want to store a million rows in memory and this value should be set.|no (default == 1000000)|
+|`maxBytesInMemory`|Long|The number of bytes to aggregate in heap memory before persisting. This is based on a rough estimate of memory usage and not actual usage. Normally this is computed internally and user does not need to set it. The maximum heap memory usage for indexing is maxBytesInMemory * (2 + maxPendingPersists).  |no (default == One-sixth of max JVM memory)|
 |`maxRowsPerSegment`|Integer|The number of rows to aggregate into a segment; this number is post-aggregation rows. Handoff will happen either if `maxRowsPerSegment` is hit or every `intermediateHandoffPeriod`, whichever happens earlier.|no (default == 5000000)|
 |`intermediatePersistPeriod`|ISO8601 Period|The period that determines the rate at which intermediate persists occur.|no (default == PT10M)|
 |`maxPendingPersists`|Integer|Maximum number of persists that can be pending but not started. If this limit would be exceeded by a new intermediate persist, ingestion will block until the currently-running persist finishes. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|no (default == 0, meaning one persist can be running concurrently with ingestion, and none can be queued up)|

docs/content/ingestion/batch-ingestion.md

@@ -154,7 +154,8 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |workingPath|String|The working path to use for intermediate results (results between Hadoop jobs).|no (default == '/tmp/druid-indexing')|
 |version|String|The version of created segments. Ignored for HadoopIndexTask unless useExplicitVersion is set to true|no (default == datetime that indexing starts at)|
 |partitionsSpec|Object|A specification of how to partition each time bucket into segments. Absence of this property means no partitioning will occur. See 'Partitioning specification' below.|no (default == 'hashed')|
-|maxRowsInMemory|Integer|The number of rows to aggregate before persisting. Note that this is the number of post-aggregation rows which may not be equal to the number of input events due to roll-up. This is used to manage the required JVM heap size.|no (default == 75000)|
+|maxRowsInMemory|Integer|The number of rows to aggregate before persisting. Note that this is the number of post-aggregation rows which may not be equal to the number of input events due to roll-up. This is used to manage the required JVM heap size. Normally user does not need to set this, but depending on the nature of data, if rows are short in terms of bytes, user may not want to store a million rows in memory and this value should be set.|no (default == 1000000)|
+|maxBytesInMemory|Long|The number of bytes to aggregate in heap memory before persisting. Normally this is computed internally and user does not need to set it. This is based on a rough estimate of memory usage and not actual usage. The maximum heap memory usage for indexing is maxBytesInMemory * (2 + maxPendingPersists).|no (default == One-sixth of max JVM memory)|
 |leaveIntermediate|Boolean|Leave behind intermediate files (for debugging) in the workingPath when a job completes, whether it passes or fails.|no (default == false)|
 |cleanupOnFailure|Boolean|Clean up intermediate files when a job fails (unless leaveIntermediate is on).|no (default == true)|
 |overwriteFiles|Boolean|Override existing files found during indexing.|no (default == false)|

docs/content/ingestion/stream-pull.md

@@ -92,7 +92,7 @@ The property `druid.realtime.specFile` has the path of a file (absolute or relat
     },
     "tuningConfig": {
       "type" : "realtime",
-      "maxRowsInMemory": 75000,
+      "maxRowsInMemory": 1000000,
       "intermediatePersistPeriod": "PT10m",
       "windowPeriod": "PT10m",
       "basePersistDirectory": "\/tmp\/realtime\/basePersist",
@@ -141,7 +141,8 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |Field|Type|Description|Required|
 |-----|----|-----------|--------|
 |type|String|This should always be 'realtime'.|no|
-|maxRowsInMemory|Integer|The number of rows to aggregate before persisting. This number is the post-aggregation rows, so it is not equivalent to the number of input events, but the number of aggregated rows that those events result in. This is used to manage the required JVM heap size. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|no (default == 75000)|
+|maxRowsInMemory|Integer|The number of rows to aggregate before persisting. This number is the post-aggregation rows, so it is not equivalent to the number of input events, but the number of aggregated rows that those events result in. This is used to manage the required JVM heap size. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|no (default == 1000000)|
+|maxBytesInMemory|Long|The maximum number of bytes to keep in memory to aggregate before persisting. This is used to manage the required JVM heap size.|no (default == One-sixth of max JVM memory)|
 |windowPeriod|ISO 8601 Period String|The amount of lag time to allow events. This is configured with a 10 minute window, meaning that any event more than 10 minutes ago will be thrown away and not included in the segment generated by the realtime server.|no (default == PT10m)|
 |intermediatePersistPeriod|ISO8601 Period String|The period that determines the rate at which intermediate persists occur. These persists determine how often commits happen against the incoming realtime stream. If the realtime data loading process is interrupted at time T, it should be restarted to re-read data that arrived at T minus this period.|no (default == PT10m)|
 |basePersistDirectory|String|The directory to put things that need persistence. The plumber is responsible for the actual intermediate persists and this tells it where to store those persists.|no (default == java tmp dir)|
@@ -287,7 +288,8 @@ The following table summarizes constraints between settings in the spec file for
 |segmentGranularity| Time granularity (minute, hour, day, week, month) for loading data at query time | equal to indexGranularity| more than queryGranularity|
 |queryGranularity| Time granularity (minute, hour, day, week, month) for rollup | less than segmentGranularity| minute, hour, day, week, month |
 |intermediatePersistPeriod| The max time (ISO8601 Period) between flushes of ingested rows from memory to disk | avoid excessive flushing | number of un-persisted rows in memory also constrained by maxRowsInMemory |
-|maxRowsInMemory| The max number of ingested rows to hold in memory before a flush to disk | number of un-persisted post-aggregation rows in memory is also constrained by intermediatePersistPeriod | use this to avoid running out of heap if too many rows in an intermediatePersistPeriod |
+|maxRowsInMemory| The max number of ingested rows to hold in memory before a flush to disk. Normally user does not need to set this, but depending on the nature of data, if rows are short in terms of bytes, user may not want to store a million rows in memory and this value should be set| number of un-persisted post-aggregation rows in memory is also constrained by intermediatePersistPeriod | use this to avoid running out of heap if too many rows in an intermediatePersistPeriod |
+|maxBytesInMemory| The number of bytes to keep in memory before a flush to disk. Normally this is computed internally and user does not need to set it. This is based on a rough estimate of memory usage and not actual usage. The maximum heap memory usage for indexing is maxBytesInMemory * (2 + maxPendingPersists)| number of un-persisted post-aggregation bytes in memory is also constrained by intermediatePersistPeriod | use this to avoid running out of heap if too many rows in an intermediatePersistPeriod |
 
 The normal, expected use cases have the following overall constraints: `intermediatePersistPeriod ≤ windowPeriod < segmentGranularity` and `queryGranularity ≤ segmentGranularity`
 

docs/content/ingestion/tasks.md

@@ -77,7 +77,7 @@ The Index Task is a simpler variation of the Index Hadoop task that is designed
     "tuningConfig" : {
       "type" : "index",
       "targetPartitionSize" : 5000000,
-      "maxRowsInMemory" : 75000
+      "maxRowsInMemory" : 1000000
     }
   }
 }
@@ -137,7 +137,8 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |--------|-----------|-------|---------|
 |type|The task type, this should always be "index".|none|yes|
 |targetPartitionSize|Used in sharding. Determines how many rows are in each segment.|5000000|no|
-|maxRowsInMemory|Used in determining when intermediate persists to disk should occur.|75000|no|
+|maxRowsInMemory|Used in determining when intermediate persists to disk should occur. Normally user does not need to set this, but depending on the nature of data, if rows are short in terms of bytes, user may not want to store a million rows in memory and this value should be set.|1000000|no|
+|maxBytesInMemory|Used in determining when intermediate persists to disk should occur. Normally this is computed internally and user does not need to set it. This value represents number of bytes to aggregate in heap memory before persisting. This is based on a rough estimate of memory usage and not actual usage. The maximum heap memory usage for indexing is maxBytesInMemory * (2 + maxPendingPersists)|1/6 of max JVM memory|no|
 |maxTotalRows|Total number of rows in segments waiting for being published. Used in determining when intermediate publish should occur.|150000|no|
 |numShards|Directly specify the number of shards to create. If this is specified and 'intervals' is specified in the granularitySpec, the index task can skip the determine intervals/partitions pass through the data. numShards cannot be specified if targetPartitionSize is set.|null|no|
 |indexSpec|defines segment storage format options to be used at indexing time, see [IndexSpec](#indexspec)|null|no|

examples/conf-quickstart/tranquility/kafka.json

@@ -53,7 +53,6 @@
         },
         "tuningConfig" : {
           "type" : "realtime",
-          "maxRowsInMemory" : "100000",
           "intermediatePersistPeriod" : "PT10M",
           "windowPeriod" : "PT10M"
         }

examples/conf-quickstart/tranquility/server.json

@@ -53,7 +53,6 @@
         },
         "tuningConfig" : {
           "type" : "realtime",
-          "maxRowsInMemory" : "100000",
           "intermediatePersistPeriod" : "PT10M",
           "windowPeriod" : "PT10M"
         }

examples/conf/tranquility/kafka.json

@@ -53,7 +53,6 @@
         },
         "tuningConfig" : {
           "type" : "realtime",
-          "maxRowsInMemory" : "100000",
           "intermediatePersistPeriod" : "PT10M",
           "windowPeriod" : "PT10M"
         }

examples/conf/tranquility/server.json

@@ -53,7 +53,6 @@
         },
         "tuningConfig" : {
           "type" : "realtime",
-          "maxRowsInMemory" : "100000",
           "intermediatePersistPeriod" : "PT10M",
           "windowPeriod" : "PT10M"
         }

extensions-contrib/orc-extensions/src/test/java/io/druid/data/input/orc/OrcIndexGeneratorJobTest.java

@@ -222,6 +222,7 @@ public void setUp() throws Exception
                 null,
                 null,
                 null,
+                null,
                 false,
                 false,
                 false,

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaTuningConfig.java

@@ -38,6 +38,7 @@ public class KafkaTuningConfig implements TuningConfig, AppenderatorConfig
   private static final boolean DEFAULT_RESET_OFFSET_AUTOMATICALLY = false;
 
   private final int maxRowsInMemory;
+  private final long maxBytesInMemory;
   private final int maxRowsPerSegment;
   private final Period intermediatePersistPeriod;
   private final File basePersistDirectory;
@@ -58,6 +59,7 @@ public class KafkaTuningConfig implements TuningConfig, AppenderatorConfig
   @JsonCreator
   public KafkaTuningConfig(
       @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory,
+      @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory,
       @JsonProperty("maxRowsPerSegment") @Nullable Integer maxRowsPerSegment,
       @JsonProperty("intermediatePersistPeriod") @Nullable Period intermediatePersistPeriod,
       @JsonProperty("basePersistDirectory") @Nullable File basePersistDirectory,
@@ -80,6 +82,9 @@ public KafkaTuningConfig(
 
     this.maxRowsInMemory = maxRowsInMemory == null ? defaults.getMaxRowsInMemory() : maxRowsInMemory;
     this.maxRowsPerSegment = maxRowsPerSegment == null ? DEFAULT_MAX_ROWS_PER_SEGMENT : maxRowsPerSegment;
+    // initializing this to 0, it will be lazily initialized to a value
+    // @see server.src.main.java.io.druid.segment.indexing.TuningConfigs#getMaxBytesInMemoryOrDefault(long)
+    this.maxBytesInMemory = maxBytesInMemory == null ? 0 : maxBytesInMemory;
     this.intermediatePersistPeriod = intermediatePersistPeriod == null
                                      ? defaults.getIntermediatePersistPeriod()
                                      : intermediatePersistPeriod;
@@ -116,6 +121,7 @@ public static KafkaTuningConfig copyOf(KafkaTuningConfig config)
   {
     return new KafkaTuningConfig(
         config.maxRowsInMemory,
+        config.maxBytesInMemory,
         config.maxRowsPerSegment,
         config.intermediatePersistPeriod,
         config.basePersistDirectory,
@@ -140,6 +146,13 @@ public int getMaxRowsInMemory()
     return maxRowsInMemory;
   }
 
+  @Override
+  @JsonProperty
+  public long getMaxBytesInMemory()
+  {
+    return maxBytesInMemory;
+  }
+
   @JsonProperty
   public int getMaxRowsPerSegment()
   {
@@ -240,6 +253,7 @@ public KafkaTuningConfig withBasePersistDirectory(File dir)
   {
     return new KafkaTuningConfig(
         maxRowsInMemory,
+        maxBytesInMemory,
         maxRowsPerSegment,
         intermediatePersistPeriod,
         dir,
@@ -269,6 +283,7 @@ public boolean equals(Object o)
     KafkaTuningConfig that = (KafkaTuningConfig) o;
     return maxRowsInMemory == that.maxRowsInMemory &&
            maxRowsPerSegment == that.maxRowsPerSegment &&
+           maxBytesInMemory == that.maxBytesInMemory &&
            maxPendingPersists == that.maxPendingPersists &&
            reportParseExceptions == that.reportParseExceptions &&
            handoffConditionTimeout == that.handoffConditionTimeout &&
@@ -289,6 +304,7 @@ public int hashCode()
     return Objects.hash(
         maxRowsInMemory,
         maxRowsPerSegment,
+        maxBytesInMemory,
         intermediatePersistPeriod,
         basePersistDirectory,
         maxPendingPersists,
@@ -310,6 +326,7 @@ public String toString()
     return "KafkaTuningConfig{" +
            "maxRowsInMemory=" + maxRowsInMemory +
            ", maxRowsPerSegment=" + maxRowsPerSegment +
+           ", maxBytesInMemory=" + maxBytesInMemory +
            ", intermediatePersistPeriod=" + intermediatePersistPeriod +
            ", basePersistDirectory=" + basePersistDirectory +
            ", maxPendingPersists=" + maxPendingPersists +

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java

@@ -93,6 +93,7 @@ public KafkaSupervisorSpec(
                             null,
                             null,
                             null,
+                            null,
                             null
                         );
     this.ioConfig = Preconditions.checkNotNull(ioConfig, "ioConfig");

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.druid.indexing.kafka.KafkaTuningConfig;
+import io.druid.segment.indexing.TuningConfigs;
 import io.druid.segment.writeout.SegmentWriteOutMediumFactory;
 import io.druid.segment.IndexSpec;
 import org.joda.time.Duration;
@@ -40,6 +41,7 @@ public class KafkaSupervisorTuningConfig extends KafkaTuningConfig
 
   public KafkaSupervisorTuningConfig(
       @JsonProperty("maxRowsInMemory") Integer maxRowsInMemory,
+      @JsonProperty("maxBytesInMemory") Long maxBytesInMemory,
       @JsonProperty("maxRowsPerSegment") Integer maxRowsPerSegment,
       @JsonProperty("intermediatePersistPeriod") Period intermediatePersistPeriod,
       @JsonProperty("basePersistDirectory") File basePersistDirectory,
@@ -65,6 +67,7 @@ public KafkaSupervisorTuningConfig(
   {
     super(
         maxRowsInMemory,
+        maxBytesInMemory,
         maxRowsPerSegment,
         intermediatePersistPeriod,
         basePersistDirectory,
@@ -131,6 +134,7 @@ public String toString()
     return "KafkaSupervisorTuningConfig{" +
            "maxRowsInMemory=" + getMaxRowsInMemory() +
            ", maxRowsPerSegment=" + getMaxRowsPerSegment() +
+           ", maxBytesInMemory=" + TuningConfigs.getMaxBytesInMemoryOrDefault(getMaxBytesInMemory()) +
            ", intermediatePersistPeriod=" + getIntermediatePersistPeriod() +
            ", basePersistDirectory=" + getBasePersistDirectory() +
            ", maxPendingPersists=" + getMaxPendingPersists() +

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -1965,6 +1965,7 @@ private KafkaIndexTask createTask(
   {
     final KafkaTuningConfig tuningConfig = new KafkaTuningConfig(
         1000,
+        null,
         maxRowsPerSegment,
         new Period("P1Y"),
         null,
@@ -2007,6 +2008,7 @@ private KafkaIndexTask createTask(
   {
     final KafkaTuningConfig tuningConfig = new KafkaTuningConfig(
         1000,
+        null,
         maxRowsPerSegment,
         new Period("P1Y"),
         null,

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaTuningConfigTest.java

@@ -56,7 +56,7 @@ public void testSerdeWithDefaults() throws Exception
     );
 
     Assert.assertNotNull(config.getBasePersistDirectory());
-    Assert.assertEquals(75000, config.getMaxRowsInMemory());
+    Assert.assertEquals(1000000, config.getMaxRowsInMemory());
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());
@@ -103,6 +103,7 @@ public void testCopyOf()
   {
     KafkaTuningConfig original = new KafkaTuningConfig(
         1,
+        null,
         2,
         new Period("PT3S"),
         new File("/tmp/xxx"),

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

@@ -185,6 +185,7 @@ public void setupTest()
 
     tuningConfig = new KafkaSupervisorTuningConfig(
         1000,
+        null,
         50000,
         new Period("P1Y"),
         new File("/test"),

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfigTest.java

@@ -58,7 +58,7 @@ public void testSerdeWithDefaults() throws Exception
     );
 
     Assert.assertNotNull(config.getBasePersistDirectory());
-    Assert.assertEquals(75000, config.getMaxRowsInMemory());
+    Assert.assertEquals(1000000, config.getMaxRowsInMemory());
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());