This repository has been archived by the owner on Jan 20, 2022. It is now read-only.
Feature/summingbird config #339
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| Copyright 2013 Twitter, Inc. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package com.twitter.summingbird.chill | ||
| import com.twitter.summingbird.{MutableStringConfig, SummingbirdConfig} | ||
| import com.twitter.chill.{ScalaKryoInstantiator, IKryoRegistrar, Kryo, toRich} | ||
| import com.twitter.chill.java.IterableRegistrar | ||
| import com.twitter.chill._ | ||
| import com.twitter.chill.config.{ ConfiguredInstantiator => ConfInst } | ||
| import com.twitter.summingbird.batch.{BatchID, Timestamp} | ||
|
|
||
| object SBChillRegistrar { | ||
| def kryoRegClass(clazz: Class[_]*) = | ||
| {k: Kryo => | ||
| clazz | ||
| .filter(k.alreadyRegistered(_)) | ||
| .foreach(k.register(_)) | ||
| } | ||
|
|
||
| def apply(cfg: SummingbirdConfig, iterableRegistrars: List[IKryoRegistrar]): SummingbirdConfig = { | ||
| val kryoConfig = new com.twitter.chill.config.Config with MutableStringConfig { | ||
| def summingbirdConfig = cfg | ||
| } | ||
|
|
||
| ConfInst.setSerialized( | ||
| kryoConfig, | ||
| classOf[ScalaKryoInstantiator], | ||
| new ScalaKryoInstantiator() | ||
| .withRegistrar(new IterableRegistrar(iterableRegistrars)) | ||
| .withRegistrar(kryoRegClass(classOf[BatchID], classOf[Timestamp])) | ||
| ) | ||
| kryoConfig.unwrap | ||
| } | ||
| } | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| /* | ||
| Copyright 2013 Twitter, Inc. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package com.twitter.summingbird | ||
|
|
||
| trait SummingbirdConfig { self => | ||
| def get(key: String): Option[AnyRef] | ||
| def put(key: String, v: AnyRef): SummingbirdConfig | ||
| final def +(kv: (String, AnyRef)) = put(kv._1, kv._2) | ||
| final def -(k: String) = remove(k) | ||
| def remove(key: String): SummingbirdConfig | ||
| def keys: Iterable[String] | ||
| def updates: Map[String, AnyRef] | ||
| def removes: Set[String] | ||
| def toMap: Map[String, AnyRef] = new Map[String, AnyRef] { | ||
| def get(k: String) = self.get(k) | ||
| def +[B1 >: AnyRef](kv: (String, B1)) = self.put(kv._1, kv._2.asInstanceOf[AnyRef]).toMap | ||
| def -(k: String) = self.-(k).toMap | ||
| def iterator = self.keys.iterator.map(k => (k, self.get(k).get)) | ||
| } | ||
| def updated(newMap: Map[String, AnyRef]): SummingbirdConfig = { | ||
| val removedKeys: Set[String] = keys.toSet -- newMap.keys | ||
| val changedOrAddedKeys = newMap.flatMap{ case (k, v) => | ||
| val oldVal = get(k) | ||
| if(oldVal != Some(v)) { | ||
| Some((k, v)) | ||
| } else None | ||
| } | ||
| val newWithoutRemoved = removedKeys.foldLeft(self)(_.remove(_)) | ||
| changedOrAddedKeys.foldLeft(newWithoutRemoved) {_ + _} | ||
| } | ||
| } | ||
|
|
||
| trait MutableStringConfig { | ||
| protected def summingbirdConfig: SummingbirdConfig | ||
| private var config = summingbirdConfig | ||
| def get(key: String) = { | ||
| assert(config != null) | ||
| config.get(key) match { | ||
| case Some(s) => s.toString | ||
| case None => null | ||
| } | ||
| } | ||
|
|
||
| def set(key: String, value: String) { | ||
| assert(config != null) | ||
| config = config.put(key, value) | ||
| } | ||
| def unwrap = config | ||
| } | ||
|
|
||
| /* | ||
| * The ReadableMap is the trait that must be implemented on the actual underlying config for the WrappingConfig. | ||
| * That is one of these should exist for an Hadoop Configuration, Storm Configuration, etc.. | ||
| */ | ||
| trait ReadableMap { | ||
|
There was a problem hiding this comment. If it's immutable, why call this out directly? docstring? Also, should SummingbirdConfig extend at least this trait (if not Scala's map trait)? There was a problem hiding this comment. This trait is purely for its downstream configs. Type signatures for keys are slightly different. The SummingbirdConfig is what something like the WrappingConfig implements, the wrapping config though needs access to a ReadableMap like object (Wraps an Hadoop config/storm config) to provide the underlying store. I'll add a doc string |
||
| def get(key: String): Option[AnyRef] | ||
| def keys: Set[String] | ||
| } | ||
|
|
||
|
|
||
| object WrappingConfig { | ||
| def apply(backingConfig: ReadableMap) = new WrappingConfig( | ||
| backingConfig, | ||
| Map[String, AnyRef](), | ||
| Set[String]()) | ||
| } | ||
|
|
||
| case class WrappingConfig(private val backingConfig: ReadableMap, | ||
| updates: Map[String, AnyRef], | ||
| removes: Set[String]) extends SummingbirdConfig { | ||
|
|
||
| def get(key: String) = { | ||
| updates.get(key) match { | ||
| case s@Some(_) => s | ||
| case None => | ||
| if(removes.contains(key)) | ||
| None | ||
| else | ||
| backingConfig.get(key) | ||
| } | ||
| } | ||
|
|
||
| def put(key: String, v: AnyRef): WrappingConfig = { | ||
| assert(v != null) | ||
| this.copy(updates = (updates + (key -> v)), removes = (removes - key)) | ||
| } | ||
|
|
||
| def remove(key: String): WrappingConfig = { | ||
| this.copy(updates = (updates - key), removes = (removes + key)) | ||
| } | ||
|
|
||
| def keys: Iterable[String] = { | ||
| ((backingConfig.keys ++ updates.keys) -- removes) | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should this extend MapLike, and be an actual Map itself?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oscar didn't want this to be a map, something about a bunch of functionality it doesn't need/shouldn't have. Its not a million miles away from having an immutable map interface though I agree