You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by cr...@apache.org on 2014/02/21 19:42:17 UTC
git commit: SAMZA-47;
split leveldb and l1 object cache configs so they each have their own
config value.
Repository: incubator-samza
Updated Branches:
refs/heads/master 5cdbd02b8 -> 1852e6149
SAMZA-47; split leveldb and l1 object cache configs so they each have their own config value.
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/1852e614
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/1852e614
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/1852e614
Branch: refs/heads/master
Commit: 1852e6149eb1c9c9f96d9b1e5dd10020279224ae
Parents: 5cdbd02
Author: Martin Kleppmann <ma...@rapportive.com>
Authored: Fri Feb 21 10:42:10 2014 -0800
Committer: Chris Riccomini <cr...@criccomi-mn.linkedin.biz>
Committed: Fri Feb 21 10:42:10 2014 -0800
----------------------------------------------------------------------
.../0.7.0/container/state-management.md | 34 ++++++++++----------
.../samza/container/SamzaContainerContext.java | 28 ++++++++++++++++
.../samza/storage/StorageEngineFactory.java | 8 ++---
.../apache/samza/container/SamzaContainer.scala | 7 ++--
.../kv/KeyValueStorageEngineFactory.scala | 14 ++++----
.../samza/storage/kv/LevelDbKeyValueStore.scala | 27 ++++++++++++----
.../samza/storage/kv/TestKeyValueStores.scala | 2 +-
7 files changed, 84 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/docs/learn/documentation/0.7.0/container/state-management.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/0.7.0/container/state-management.md b/docs/learn/documentation/0.7.0/container/state-management.md
index cafb0f8..c7b25b0 100644
--- a/docs/learn/documentation/0.7.0/container/state-management.md
+++ b/docs/learn/documentation/0.7.0/container/state-management.md
@@ -172,17 +172,15 @@ In this section we will give a quick tutorial on configuring and using the key-v
To declare a new store for usage you add the following to your job config:
# Use the key-value store implementation for a store called "my-store"
- stores.my-store.factory=samza.storage.kv.KeyValueStorageEngineFactory
+ stores.my-store.factory=org.apache.samza.storage.kv.KeyValueStorageEngineFactory
# Log changes to the store to an output stream for restore
# If no changelog is specified the store will not be logged (but you can still rebuild off your input streams)
- stores.my-store.changelog=my-stream-name
-
- # The system to use for the changelog stream
- stores.my-store.system=kafka
+ stores.my-store.changelog=kafka.my-stream-name
# The serialization format to use
- stores.my-store.serde=string
+ stores.my-store.key.serde=string
+ stores.my-store.msg.serde=string
Here is some simple example code that only writes to the store:
@@ -213,25 +211,27 @@ Here is the complete key-value store API:
Here is a list of additional configurations accepted by the key-value store along with their default values:
# The number of writes to batch together
- stores.my-store.batch.size=500
+ stores.my-store.write.batch.size=500
# The total number of objects to cache in the "L1" object cache. This must be at least as large as the batch.size.
# A cache size of 0 disables all caching and batching.
stores.my-store.object.cache.size=1000
- # The size of the off-heap leveldb block cache in bytes
- stores.my-store.leveldb.block.cache.size=16777216
-
- # Enable block compression using snappy?
- stores.my-store.leveldb.compress=true
+ # The size of the off-heap leveldb block cache in bytes, per container. If you have multiple tasks within
+ # one container, each task is given a proportional share of this cache.
+ stores.my-store.container.cache.size.bytes=104857600
- # The remaining options are a bit low-level and likely you don't need to change them unless you are a compulsive fiddler
+ # The amount of memory leveldb uses for buffering writes before they are written to disk, per container.
+ # If you have multiple tasks within one container, each task is given a proportional share of this buffer.
+ # This setting also determines the size of leveldb's segment files.
+ stores.my-store.container.write.buffer.size.bytes=33554432
- # The leveldb block size (see leveldb docs).
- stores.my-store.leveldb.block.size=4096
+ # Enable block compression? (set compression=none to disable)
+ stores.my-store.leveldb.compression=snappy
- # The amount of memory leveldb uses for buffering the latest segment, also the size of leveldb's segment files.
- stores.my-store.leveldb.write.buffer.size=16777216
+ # If compression is enabled, leveldb groups approximately this many uncompressed bytes into one compressed block.
+ # You probably don't need to change this unless you are a compulsive fiddler.
+ stores.my-store.leveldb.block.size.bytes=4096
### Implementing common use cases with the key-value store
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-api/src/main/java/org/apache/samza/container/SamzaContainerContext.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/container/SamzaContainerContext.java b/samza-api/src/main/java/org/apache/samza/container/SamzaContainerContext.java
new file mode 100644
index 0000000..baa0ea8
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/container/SamzaContainerContext.java
@@ -0,0 +1,28 @@
+package org.apache.samza.container;
+
+import java.util.Collection;
+
+import org.apache.samza.Partition;
+import org.apache.samza.config.Config;
+
+public class SamzaContainerContext {
+ public final String name;
+ public final Config config;
+ public final Collection<Partition> partitions;
+
+ /**
+ * An immutable context object that can passed to tasks to give them information
+ * about the container in which they are executing.
+ * @param name The name of the container (either a YARN AM or SamzaContainer).
+ * @param config The job configuration.
+ * @param partitions The set of input partitions assigned to this container.
+ */
+ public SamzaContainerContext(
+ String name,
+ Config config,
+ Collection<Partition> partitions) {
+ this.name = name;
+ this.config = config;
+ this.partitions = partitions;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java b/samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java
index fe83c8e..da57bf0 100644
--- a/samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java
+++ b/samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java
@@ -21,7 +21,7 @@ package org.apache.samza.storage;
import java.io.File;
-import org.apache.samza.config.Config;
+import org.apache.samza.container.SamzaContainerContext;
import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.serializers.Serde;
import org.apache.samza.system.SystemStreamPartition;
@@ -41,8 +41,8 @@ public interface StorageEngineFactory<K, V> {
* @param storeDir The directory of the storage engine.
* @param keySerde The serializer to use for serializing keys when reading or writing to the store.
* @param msgSerde The serializer to use for serializing messages when reading or writing to the store.
- * @param config The configuration for the task.
* @param changeLogSystemStreamPartition Samza stream partition from which to receive the changelog from.
+ * @param containerContext Information about the container in which the task is executing.
* @return The storage engine instance.
*/
public StorageEngine getStorageEngine(
@@ -51,7 +51,7 @@ public interface StorageEngineFactory<K, V> {
Serde<K> keySerde,
Serde<V> msgSerde,
MessageCollector collector,
- Config config,
MetricsRegistry registry,
- SystemStreamPartition changeLogSystemStreamPartition);
+ SystemStreamPartition changeLogSystemStreamPartition,
+ SamzaContainerContext containerContext);
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 13421d2..84a413c 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -59,6 +59,7 @@ import org.apache.samza.system.SystemConsumersMetrics
import org.apache.samza.metrics.MetricsRegistryMap
import org.apache.samza.system.chooser.DefaultChooser
import org.apache.samza.system.chooser.RoundRobinChooserFactory
+import scala.collection.JavaConversions._
object SamzaContainer extends Logging {
def main(args: Array[String]) {
@@ -333,6 +334,8 @@ object SamzaContainer extends Logging {
val partitions = inputStreams.map(_.getPartition).toSet
+ val containerContext = new SamzaContainerContext(containerName, config, partitions)
+
val taskInstances = partitions.map(partition => {
debug("Setting up task instance: %s" format partition)
@@ -376,9 +379,9 @@ object SamzaContainer extends Logging {
keySerde,
msgSerde,
collector,
- config,
taskInstanceMetrics.registry,
- changeLogSystemStreamPartition)
+ changeLogSystemStreamPartition,
+ containerContext)
(storeName, storageEngine)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngineFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngineFactory.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngineFactory.scala
index ea75fab..81fe861 100644
--- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngineFactory.scala
+++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngineFactory.scala
@@ -21,6 +21,7 @@ package org.apache.samza.storage.kv
import java.io.File
import org.apache.samza.config.Config
+import org.apache.samza.container.SamzaContainerContext
import org.apache.samza.serializers._
import org.apache.samza.SamzaException
import org.apache.samza.metrics.MetricsRegistry
@@ -36,13 +37,13 @@ class KeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] {
keySerde: Serde[K],
msgSerde: Serde[V],
collector: MessageCollector,
- config: Config,
registry: MetricsRegistry,
- changeLogSystemStreamPartition: SystemStreamPartition): StorageEngine = {
+ changeLogSystemStreamPartition: SystemStreamPartition,
+ containerContext: SamzaContainerContext): StorageEngine = {
- val storageConfig = config.subset("stores." + storeName + ".", true)
- val batchSize = config.getInt("batch.size", 500)
- val cacheSize = config.getInt("cache.size", math.max(batchSize, 1000))
+ val storageConfig = containerContext.config.subset("stores." + storeName + ".", true)
+ val batchSize = storageConfig.getInt("write.batch.size", 500)
+ val cacheSize = storageConfig.getInt("object.cache.size", math.max(batchSize, 1000))
val enableCache = cacheSize > 0
if (cacheSize > 0 && cacheSize < batchSize) {
@@ -58,7 +59,8 @@ class KeyValueStorageEngineFactory[K, V] extends StorageEngineFactory[K, V] {
}
val levelDbMetrics = new LevelDbKeyValueStoreMetrics(storeName, registry)
- val levelDb = new LevelDbKeyValueStore(storeDir, LevelDbKeyValueStore.options(storageConfig), levelDbMetrics)
+ val levelDbOptions = LevelDbKeyValueStore.options(storageConfig, containerContext)
+ val levelDb = new LevelDbKeyValueStore(storeDir, levelDbOptions, levelDbMetrics)
val maybeLoggedStore = if (changeLogSystemStreamPartition == null) {
levelDb
} else {
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-kv/src/main/scala/org/apache/samza/storage/kv/LevelDbKeyValueStore.scala
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LevelDbKeyValueStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LevelDbKeyValueStore.scala
index 4c14aa2..8602a32 100644
--- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/LevelDbKeyValueStore.scala
+++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/LevelDbKeyValueStore.scala
@@ -27,17 +27,32 @@ import java.io._
import java.util.Iterator
import java.lang.Iterable
import org.apache.samza.config.Config
-import grizzled.slf4j.Logging
+import org.apache.samza.container.SamzaContainerContext
+import grizzled.slf4j.{Logger, Logging}
object LevelDbKeyValueStore {
- def options(config: Config) = {
+ private lazy val logger = Logger(classOf[LevelDbKeyValueStore])
+
+ def options(storeConfig: Config, containerContext: SamzaContainerContext) = {
+ val cacheSize = storeConfig.getLong("container.cache.size.bytes", 100 * 1024 * 1024L)
+ val writeBufSize = storeConfig.getLong("container.write.buffer.size.bytes", 32 * 1024 * 1024)
val options = new Options
- options.blockSize(config.getInt("block.size", 4086))
- options.cacheSize(config.getLong("cache.size", 64 * 1024 * 1024L))
- options.compressionType(if (config.getBoolean("compress", true)) CompressionType.SNAPPY else CompressionType.NONE)
+
+ // Cache size and write buffer size are specified on a per-container basis.
+ options.cacheSize(cacheSize / containerContext.partitions.size)
+ options.writeBufferSize((writeBufSize / containerContext.partitions.size).toInt)
+ options.blockSize(storeConfig.getInt("leveldb.block.size.bytes", 4096))
+ options.compressionType(
+ storeConfig.get("leveldb.compression", "snappy") match {
+ case "snappy" => CompressionType.SNAPPY
+ case "none" => CompressionType.NONE
+ case _ =>
+ logger.warn("Unknown leveldb.compression codec %s, defaulting to Snappy" format storeConfig.get("leveldb.compression", "snappy"))
+ CompressionType.SNAPPY
+ }
+ )
options.createIfMissing(true)
options.errorIfExists(true)
- options.writeBufferSize(config.getInt("write.buffer.size", 16 * 1024 * 1024))
options
}
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/1852e614/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStores.scala
----------------------------------------------------------------------
diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStores.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStores.scala
index e99d459..163f68e 100644
--- a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStores.scala
+++ b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStores.scala
@@ -313,5 +313,5 @@ object TestKeyValueStores {
val CacheSize = 10
val BatchSize = 5
@Parameters
- def parameters: java.util.Collection[Array[String]] = Arrays.asList(Array("cache"), Array("serde"), Array("cache-and-serde"), Array("levledb"))
+ def parameters: java.util.Collection[Array[String]] = Arrays.asList(Array("cache"), Array("serde"), Array("cache-and-serde"), Array("leveldb"))
}