Skip to content

[SPARK-19968][SS] Use a cached instance of KafkaProducer instead of creating one every batch. #17308

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ScrapCodes wants to merge 13 commits into from
Closed

[SPARK-19968][SS] Use a cached instance of KafkaProducer instead of creating one every batch. #17308

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
8224596
[SPARK-19968][SS] Use a cached instance of KafkaProducer instead of c…
ScrapCodes May 12, 2017
e07e77e
Self review and code style improvements.
ScrapCodes May 15, 2017
d6e4088
[SPARK-20737] Mechanism for cleanup hooks, for structured-streaming s…
ScrapCodes May 17, 2017
3ec9981
Synchronized access to local hash map.
ScrapCodes May 17, 2017
c614bc0
Style feedback
ScrapCodes May 22, 2017
e5cd1e6
Implementation using guava cache.
ScrapCodes May 22, 2017
ef2d6cd
Added shutdown hook as stop at driver is not very reliable.
ScrapCodes May 23, 2017
d2b3ecd
Corrected some style issues.
ScrapCodes May 23, 2017
15dfc80
removed a stray change.
ScrapCodes May 23, 2017
039d063
Code review, feedback.
ScrapCodes May 24, 2017
1c9f892
Using spark conf instead of System.getProperties.
ScrapCodes May 24, 2017
588fa03
Michael's feedback!
ScrapCodes May 25, 2017
a10276a
Using Loading cache, and few style related fixes.
ScrapCodes May 29, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...nal/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/CachedKafkaProducer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ private[kafka010] object CachedKafkaProducer extends Logging {
private type Producer = KafkaProducer[Array[Byte], Array[Byte]]

private val cacheExpireTimeout: Long =
System.getProperty("spark.kafka.guava.cache.timeout", "10").toLong
System.getProperty("spark.kafka.guava.cache.timeout.minutes", "10").toLong
Copy link
Contributor

@brkyvz brkyvz May 24, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

don't we need to get this from SparkEnv by the way? I don't know if the properties get populated properly.
Also, adding minutes to the conf makes it kinda long right? I think we can also replace guava with producer.
I think it may also be better to use this so that we get rid of minutes and users can actually provide arbitrary durations (hours if they want). I think that's what we generally use for duration type confs.

Copy link
Member Author

@ScrapCodes ScrapCodes May 24, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, you are right !


private val removalListener = new RemovalListener[String, Producer]() {
override def onRemoval(notification: RemovalNotification[String, Producer]): Unit = {
Expand Down
3 changes: 1 addition & 2 deletions external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package org.apache.spark.sql.kafka010

import java.{util => ju}

import org.apache.spark.SparkContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.expressions._
Expand Down Expand Up @@ -49,7 +48,7 @@ private[kafka010] object KafkaWriter extends Logging {
topic: Option[String] = None): Unit = {
val schema = queryExecution.analyzed.output
schema.find(_.name == TOPIC_ATTRIBUTE_NAME).getOrElse(
if (topic == None) {
if (topic.isEmpty) {
throw new AnalysisException(s"topic option required when no " +
s"'$TOPIC_ATTRIBUTE_NAME' attribute is present. Use the " +
s"${KafkaSourceProvider.TOPIC_OPTION_KEY} option for setting a topic.")
Expand Down