You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2019/04/06 23:05:33 UTC
[spark] branch master updated: [SPARK-27399][STREAMING][KAFKA]
Arrange scattered config and reduce hardcode for kafka 10.
This is an automated email from the ASF dual-hosted git repository.
srowen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 53e31e2 [SPARK-27399][STREAMING][KAFKA] Arrange scattered config and reduce hardcode for kafka 10.
53e31e2 is described below
commit 53e31e2ca1f0aec1d19f1528cdd1a5317a4e0a32
Author: gengjiaan <ge...@360.cn>
AuthorDate: Sat Apr 6 18:05:15 2019 -0500
[SPARK-27399][STREAMING][KAFKA] Arrange scattered config and reduce hardcode for kafka 10.
## What changes were proposed in this pull request?
I found a lot scattered config in `Kafka` streaming.I think should arrange these config in unified position.
## How was this patch tested?
No need UT.
Closes #24267 from beliefer/arrange-scattered-streaming-kafka-config.
Authored-by: gengjiaan <ge...@360.cn>
Signed-off-by: Sean Owen <se...@databricks.com>
---
.../kafka010/DirectKafkaInputDStream.scala | 3 +-
.../apache/spark/streaming/kafka010/KafkaRDD.scala | 16 +++-----
.../streaming/kafka010/PerPartitionConfig.scala | 4 +-
.../apache/spark/streaming/kafka010/package.scala | 47 +++++++++++++++++++++-
4 files changed, 55 insertions(+), 15 deletions(-)
diff --git a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala
index 224f41a..88d6d0e 100644
--- a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala
+++ b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala
@@ -229,8 +229,7 @@ private[spark] class DirectKafkaInputDStream[K, V](
val fo = currentOffsets(tp)
OffsetRange(tp.topic, tp.partition, fo, uo)
}
- val useConsumerCache = context.conf.getBoolean("spark.streaming.kafka.consumer.cache.enabled",
- true)
+ val useConsumerCache = context.conf.get(CONSUMER_CACHE_ENABLED)
val rdd = new KafkaRDD[K, V](context.sparkContext, executorKafkaParams, offsetRanges.toArray,
getPreferredHosts, useConsumerCache)
diff --git a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala
index 4513dca..bd2e7e1 100644
--- a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala
+++ b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala
@@ -24,6 +24,7 @@ import org.apache.kafka.common.TopicPartition
import org.apache.spark.{Partition, SparkContext, TaskContext}
import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.Network._
import org.apache.spark.partial.{BoundedDouble, PartialResult}
import org.apache.spark.rdd.RDD
import org.apache.spark.scheduler.ExecutorCacheTaskLocation
@@ -64,16 +65,11 @@ private[spark] class KafkaRDD[K, V](
" must be set to false for executor kafka params, else offsets may commit before processing")
// TODO is it necessary to have separate configs for initial poll time vs ongoing poll time?
- private val pollTimeout = conf.getLong("spark.streaming.kafka.consumer.poll.ms",
- conf.getTimeAsSeconds("spark.network.timeout", "120s") * 1000L)
- private val cacheInitialCapacity =
- conf.getInt("spark.streaming.kafka.consumer.cache.initialCapacity", 16)
- private val cacheMaxCapacity =
- conf.getInt("spark.streaming.kafka.consumer.cache.maxCapacity", 64)
- private val cacheLoadFactor =
- conf.getDouble("spark.streaming.kafka.consumer.cache.loadFactor", 0.75).toFloat
- private val compacted =
- conf.getBoolean("spark.streaming.kafka.allowNonConsecutiveOffsets", false)
+ private val pollTimeout = conf.get(CONSUMER_POLL_MS).getOrElse(conf.get(NETWORK_TIMEOUT) * 1000L)
+ private val cacheInitialCapacity = conf.get(CONSUMER_CACHE_INITIAL_CAPACITY)
+ private val cacheMaxCapacity = conf.get(CONSUMER_CACHE_MAX_CAPACITY)
+ private val cacheLoadFactor = conf.get(CONSUMER_CACHE_LOAD_FACTOR).toFloat
+ private val compacted = conf.get(ALLOW_NON_CONSECUTIVE_OFFSETS)
override def persist(newLevel: StorageLevel): this.type = {
logError("Kafka ConsumerRecord is not serializable. " +
diff --git a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/PerPartitionConfig.scala b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/PerPartitionConfig.scala
index 77193e2..b261500 100644
--- a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/PerPartitionConfig.scala
+++ b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/PerPartitionConfig.scala
@@ -39,8 +39,8 @@ abstract class PerPartitionConfig extends Serializable {
*/
private class DefaultPerPartitionConfig(conf: SparkConf)
extends PerPartitionConfig {
- val maxRate = conf.getLong("spark.streaming.kafka.maxRatePerPartition", 0)
- val minRate = conf.getLong("spark.streaming.kafka.minRatePerPartition", 1)
+ val maxRate = conf.get(MAX_RATE_PER_PARTITION)
+ val minRate = conf.get(MIN_RATE_PER_PARTITION)
def maxRatePerPartition(topicPartition: TopicPartition): Long = maxRate
override def minRatePerPartition(topicPartition: TopicPartition): Long = minRate
diff --git a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/package.scala b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/package.scala
index 09db6d6..3d2921f 100644
--- a/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/package.scala
+++ b/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/package.scala
@@ -17,7 +17,52 @@
package org.apache.spark.streaming
+import org.apache.spark.internal.config.ConfigBuilder
+
/**
* Spark Integration for Kafka 0.10
*/
-package object kafka010 //scalastyle:ignore
+package object kafka010 { //scalastyle:ignore
+
+ private[spark] val CONSUMER_CACHE_ENABLED =
+ ConfigBuilder("spark.streaming.kafka.consumer.cache.enabled")
+ .booleanConf
+ .createWithDefault(true)
+
+ private[spark] val CONSUMER_POLL_MS =
+ ConfigBuilder("spark.streaming.kafka.consumer.poll.ms")
+ .longConf
+ .createOptional
+
+ private[spark] val CONSUMER_CACHE_INITIAL_CAPACITY =
+ ConfigBuilder("spark.streaming.kafka.consumer.cache.initialCapacity")
+ .intConf
+ .createWithDefault(16)
+
+ private[spark] val CONSUMER_CACHE_MAX_CAPACITY =
+ ConfigBuilder("spark.streaming.kafka.consumer.cache.maxCapacity")
+ .intConf
+ .createWithDefault(64)
+
+ private[spark] val CONSUMER_CACHE_LOAD_FACTOR =
+ ConfigBuilder("spark.streaming.kafka.consumer.cache.loadFactor")
+ .doubleConf
+ .createWithDefault(0.75)
+
+ private[spark] val MAX_RATE_PER_PARTITION =
+ ConfigBuilder("spark.streaming.kafka.maxRatePerPartition")
+ .longConf
+ .createWithDefault(0)
+
+ private[spark] val MIN_RATE_PER_PARTITION =
+ ConfigBuilder("spark.streaming.kafka.minRatePerPartition")
+ .longConf
+ .createWithDefault(1)
+
+ private[spark] val ALLOW_NON_CONSECUTIVE_OFFSETS =
+ ConfigBuilder("spark.streaming.kafka.allowNonConsecutiveOffsets")
+ .booleanConf
+ .createWithDefault(false)
+
+}
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org