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

Make sure to register BatchID and Timespan #324

Merged
merged 2 commits into from
Oct 24, 2013
Merged

Make sure to register BatchID and Timespan #324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
17f1859
Make sure to register BatchID and Timespan
johnynek Oct 24, 2013
7a1178a
Merge branch 'develop' of github.com:twitter/summingbird into registe…
ianoc Oct 24, 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
10 changes: 10 additions & 0 deletions summingbird-builder/src/main/scala/com/twitter/summingbird/scalding/ScaldingEnv.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,15 @@ package com.twitter.summingbird.scalding
import com.twitter.bijection.Conversion.asMethod
import com.twitter.chill.ScalaKryoInstantiator
import com.twitter.chill.java.IterableRegistrar
import com.twitter.chill.{Kryo, IKryoRegistrar, toRich }
import com.twitter.chill.config.{ ConfiguredInstantiator => ConfInst, JavaMapConfig }
import com.twitter.scalding.{ Tool => STool, _ }
import com.twitter.summingbird.scalding.store.HDFSMetadata
import com.twitter.summingbird.{ Env, Unzip2, Summer, Producer, TailProducer, AbstractJob }
import com.twitter.summingbird.batch.{ BatchID, Batcher, Timestamp }
import com.twitter.summingbird.builder.{ SourceBuilder, Reducers, CompletedBuilder }
import com.twitter.summingbird.storm.Storm

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.util.ToolRunner
import org.apache.hadoop.util.GenericOptionsParser
Expand Down Expand Up @@ -105,6 +107,14 @@ case class ScaldingEnv(override val jobName: String, inargs: Array[String])
new ScalaKryoInstantiator()
.withRegistrar(builder.registrar)
.withRegistrar(new IterableRegistrar(ajob.registrars))
.withRegistrar(new IKryoRegistrar {
def apply(k: Kryo) {
List(classOf[BatchID], classOf[Timestamp])
.foreach { cls =>
if(!k.alreadyRegistered(cls)) k.register(cls)
}
}
})
)
fromMap(ajob.transformConfig(jConf.as[Map[String, AnyRef]]))
}
Expand Down
11 changes: 10 additions & 1 deletion summingbird-builder/src/main/scala/com/twitter/summingbird/storm/StormEnv.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,11 @@ package com.twitter.summingbird.storm

import backtype.storm.Config
import com.twitter.scalding.Args
import com.twitter.chill.ScalaKryoInstantiator
import com.twitter.chill.{ScalaKryoInstantiator, Kryo, toRich, IKryoRegistrar}
import com.twitter.chill.config.{ ConfiguredInstantiator => ConfInst, JavaMapConfig }
import com.twitter.chill.java.IterableRegistrar
import com.twitter.summingbird.{ Env, Unzip2, Producer, TailProducer }
import com.twitter.summingbird.batch.{ BatchID, Timestamp }
import com.twitter.summingbird.scalding.Scalding
import scala.collection.JavaConverters._

Expand Down Expand Up @@ -57,6 +58,14 @@ case class StormEnv(override val jobName: String, override val args: Args)
new ScalaKryoInstantiator()
.withRegistrar(builder.registrar)
.withRegistrar(new IterableRegistrar(ajob.registrars))
.withRegistrar(new IKryoRegistrar {
def apply(k: Kryo) {
List(classOf[BatchID], classOf[Timestamp])
.foreach { cls =>
if(!k.alreadyRegistered(cls)) k.register(cls)
}
}
})
)
transformed
}.run(
Expand Down