Skip to content
This repository was archived by the owner on Jan 20, 2022. It is now read-only.

Add local or shuffle option #384

Merged
johnynek merged 3 commits into from
Nov 27, 2013
Merged

Add local or shuffle option #384

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8c928ee
Add local or shuffle option
ianoc Nov 27, 2013
bf939ee
Rename away from storm specific naming
ianoc Nov 27, 2013
4fa45f9
Dependency instead of dependant
ianoc Nov 27, 2013
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
1 change: 1 addition & 0 deletions summingbird-storm/src/main/scala/com/twitter/summingbird/storm/Constants.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,4 +44,5 @@ object Constants {
val DEFAULT_MONOID_IS_COMMUTATIVE = MonoidIsCommutative.default
val DEFAULT_MAX_WAITING_FUTURES = MaxWaitingFutures(10)
val DEFAULT_MAX_FUTURE_WAIT_TIME = MaxFutureWaitTime(Duration.fromSeconds(60))
val DEFAULT_FM_PREFER_LOCAL_DEPENDENCY = PreferLocalDependency(false)
}
12 changes: 9 additions & 3 deletions summingbird-storm/src/main/scala/com/twitter/summingbird/storm/StormPlatform.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,9 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird
val maxWaitTime = getOrElse(stormDag, node, DEFAULT_MAX_FUTURE_WAIT_TIME)
logger.info("[{}] maxWaiting: {}", nodeName, maxWaiting.get)

val usePreferLocalDependency = getOrElse(stormDag, node, DEFAULT_FM_PREFER_LOCAL_DEPENDENCY)
logger.info("[{}] usePreferLocalDependency: {}", nodeName, usePreferLocalDependency.get)

val summerOpt:Option[SummerNode[Storm]] = stormDag.dependantsOf(node).collect{case s: SummerNode[Storm] => s}.headOption

val bolt = summerOpt match {
Expand Down Expand Up @@ -214,9 +217,12 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird


val dependenciesNames = stormDag.dependenciesOf(node).collect { case x: StormNode => stormDag.getNodeName(x) }
// TODO: https://github.com/twitter/summingbird/issues/366
// test localOrShuffleGrouping here. may give big wins for serialization heavy jobs.
dependenciesNames.foreach { declarer.shuffleGrouping(_) }
if (usePreferLocalDependency.get) {
dependenciesNames.foreach { declarer.localOrShuffleGrouping(_) }
} else {
dependenciesNames.foreach { declarer.shuffleGrouping(_) }
}

}

private def scheduleSpout[K](stormDag: Dag[Storm], node: StormNode)(implicit topologyBuilder: TopologyBuilder) = {
Expand Down
4 changes: 3 additions & 1 deletion summingbird-storm/src/main/scala/com/twitter/summingbird/storm/option/FlatMapOptions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,4 +56,6 @@ object SpoutStormMetrics {
class SpoutStormMetrics(val metrics: () => TraversableOnce[StormMetric[IMetric]]) {
def toSpoutMetrics: () => TraversableOnce[Metric[IMetric]] =
{() => metrics().map{ x: StormMetric[IMetric] => Metric(x.name, x.metric, x.interval.inSeconds)}}
}
}

case class PreferLocalDependency(get: Boolean)