You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by su...@apache.org on 2020/12/05 00:40:04 UTC
[incubator-gobblin] branch master updated: [GOBBLIN-1331] Add
average record size to work unit state[]
This is an automated email from the ASF dual-hosted git repository.
suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 2d2b948 [GOBBLIN-1331] Add average record size to work unit state[]
2d2b948 is described below
commit 2d2b948a4ae9cbb47caae39327f9ad9e2b476893
Author: suvasude <su...@linkedin.biz>
AuthorDate: Fri Dec 4 16:39:58 2020 -0800
[GOBBLIN-1331] Add average record size to work unit state[]
Closes #3166 from sv2000/avgRecordSize
---
.../packer/KafkaTopicGroupingWorkUnitPacker.java | 25 +++++++++++++---------
1 file changed, 15 insertions(+), 10 deletions(-)
diff --git a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/workunit/packer/KafkaTopicGroupingWorkUnitPacker.java b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/workunit/packer/KafkaTopicGroupingWorkUnitPacker.java
index aeab8e0..ba5b608 100644
--- a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/workunit/packer/KafkaTopicGroupingWorkUnitPacker.java
+++ b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/workunit/packer/KafkaTopicGroupingWorkUnitPacker.java
@@ -68,6 +68,7 @@ import static org.apache.gobblin.source.extractor.extract.kafka.workunit.packer.
*/
@Slf4j
public class KafkaTopicGroupingWorkUnitPacker extends KafkaWorkUnitPacker {
+ public static final String GOBBLIN_KAFKA_PREFIX = "gobblin.kafka.";
private static final int DEFAULT_NUM_TOPIC_PARTITIONS_PER_CONTAINER = 10;
private static final String TOPIC_PARTITION_DELIMITER = "-";
@@ -76,24 +77,25 @@ public class KafkaTopicGroupingWorkUnitPacker extends KafkaWorkUnitPacker {
// packed into a single workunit. For example, if the container capacity is set to 10, and each topic partition has a
// weight of 1, then 10 partitions of the topic will be packed into a single workunit. This configuration is topic-independent
// i.e. all topics will be assumed to have the same peak consumption rate when set.
- public static final String CONTAINER_CAPACITY_KEY = "gobblin.kafka.streaming.containerCapacity";
+ public static final String CONTAINER_CAPACITY_KEY = GOBBLIN_KAFKA_PREFIX + "streaming.containerCapacity";
public static final double DEFAULT_CONTAINER_CAPACITY = 10;
//A boolean flag to enable per-topic container capacity, where "container capacity" is as defined earlier. This
// configuration is useful in scenarios where the write performance can vary significantly across topics due to differences
// in schema, as in the case of columnar formats such as ORC and Parquet. When enabled, the bin packing algorithm uses
// historic consumption rates for a given topic as tracked by the ingestion pipeline.
-
- public static final String IS_PER_TOPIC_CONTAINER_CAPACITY_ENABLED_KEY = "gobblin.kafka.streaming.isPerTopicBinCapacityEnabled";
+ public static final String IS_PER_TOPIC_CONTAINER_CAPACITY_ENABLED_KEY = GOBBLIN_KAFKA_PREFIX + "streaming.isPerTopicBinCapacityEnabled";
public static final Boolean DEFAULT_IS_PER_TOPIC_CONTAINER_CAPACITY_ENABLED = false;
//A topic-specific config that controls the minimum number of containers for that topic.
- public static final String MIN_CONTAINERS_FOR_TOPIC = "gobblin.kafka.minContainersForTopic";
- public static final String PARTITION_WATERMARK = "gobblin.kafka.partition.watermark";
- public static final String PACKING_START_TIME_MILLIS = "gobblin.kafka.packer.packingStartTimeMillis";
- public static final String IS_STATS_BASED_PACKING_ENABLED_KEY = "gobblin.kafka.streaming.isStatsBasedPackingEnabled";
+ public static final String MIN_CONTAINERS_FOR_TOPIC = GOBBLIN_KAFKA_PREFIX + "minContainersForTopic";
+ public static final String PARTITION_WATERMARK = GOBBLIN_KAFKA_PREFIX + "partition.watermark";
+
+ public static final String PACKING_START_TIME_MILLIS = GOBBLIN_KAFKA_PREFIX + "packer.packingStartTimeMillis";
+ public static final String IS_STATS_BASED_PACKING_ENABLED_KEY = GOBBLIN_KAFKA_PREFIX + "streaming.isStatsBasedPackingEnabled";
public static final boolean DEFAULT_IS_STATS_BASED_PACKING_ENABLED = false;
- public static final String CONTAINER_CAPACITY_COMPUTATION_STRATEGY_KEY = "gobblin.kafka.streaming.containerCapacityComputationStrategy";
+ public static final String CONTAINER_CAPACITY_COMPUTATION_STRATEGY_KEY =
+ GOBBLIN_KAFKA_PREFIX + "streaming.containerCapacityComputationStrategy";
public static final String DEFAULT_CONTAINER_CAPACITY_COMPUTATION_STRATEGY = ContainerCapacityComputationStrategy.MEDIAN.name();
public enum ContainerCapacityComputationStrategy {
@@ -113,9 +115,7 @@ public class KafkaTopicGroupingWorkUnitPacker extends KafkaWorkUnitPacker {
/**
* When indexing-packing is enabled, the number of partitions is important for extractor to know
* how many kafka partitions need to be pulled from.
- *
* Set to public-static to share with Extractor.
- * TODO: Shall be changed to package-private
*/
public static final String NUM_PARTITIONS_ASSIGNED = "gobblin.kafka.streaming.numPartitions";
//A derived metric that defines the default workunit size, in case of workunit size cannot be estimated.
@@ -257,6 +257,11 @@ public class KafkaTopicGroupingWorkUnitPacker extends KafkaWorkUnitPacker {
workUnit.setProp(PACKING_START_TIME_MILLIS, this.packingStartTimeMillis);
workUnit.setProp(DEFAULT_WORKUNIT_SIZE_KEY, getDefaultWorkUnitSize());
workUnit.setProp(MIN_WORKUNIT_SIZE_KEY, getMinWorkUnitSize(workUnit));
+ // avgRecordSize is unknown when bootstrapping. so skipping setting this
+ // and ORC writer will use the default setting for the tunning feature.
+ if (watermark != null && watermark.getAvgRecordSize() > 0) {
+ workUnit.setProp(ConfigurationKeys.AVG_RECORD_SIZE, watermark.getAvgRecordSize());
+ }
}
}
}