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/10/27 20:20:06 UTC
git commit: SAMZA-434; rocksdb docs
Repository: incubator-samza
Updated Branches:
refs/heads/0.8.0 147d18e01 -> 0bab52d82
SAMZA-434; rocksdb docs
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/0bab52d8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/0bab52d8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/0bab52d8
Branch: refs/heads/0.8.0
Commit: 0bab52d828a915dea7b5a6a9910efde0846ba16e
Parents: 147d18e
Author: Chris Riccomini <cr...@criccomi-mn.linkedin.biz>
Authored: Mon Oct 27 12:17:45 2014 -0700
Committer: Chris Riccomini <cr...@criccomi-mn.linkedin.biz>
Committed: Mon Oct 27 12:19:51 2014 -0700
----------------------------------------------------------------------
.../versioned/container/state-management.md | 20 ++-
.../versioned/jobs/configuration-table.html | 158 +++++++++++++++++--
.../RocksDbKeyValueStorageEngineFactory.scala | 2 +-
3 files changed, 159 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/0bab52d8/docs/learn/documentation/versioned/container/state-management.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/container/state-management.md b/docs/learn/documentation/versioned/container/state-management.md
index d71de0b..79067bb 100644
--- a/docs/learn/documentation/versioned/container/state-management.md
+++ b/docs/learn/documentation/versioned/container/state-management.md
@@ -126,17 +126,17 @@ Nothing prevents you from using an external database if you want to, but for man
### Key-value storage
-Any storage engine can be plugged into Samza, as described below. Out of the box, Samza ships with a key-value store implementation that is built on [LevelDB](https://code.google.com/p/leveldb) using a [JNI API](https://github.com/fusesource/leveldbjni).
+Any storage engine can be plugged into Samza, as described below. Out of the box, Samza ships with a key-value store implementation that is built on [RocksDB](http://rocksdb.org) using a [JNI API](https://github.com/facebook/rocksdb/wiki/RocksJava-Basics).
-LevelDB has several nice properties. Its memory allocation is outside of the Java heap, which makes it more memory-efficient and less prone to garbage collection pauses than a Java-based storage engine. It is very fast for small datasets that fit in memory; datasets larger than memory are slower but still possible. It is [log-structured](http://www.igvita.com/2012/02/06/sstable-and-log-structured-storage-leveldb/), allowing very fast writes. It also includes support for block compression, which helps to reduce I/O and memory usage.
+RocksDB has several nice properties. Its memory allocation is outside of the Java heap, which makes it more memory-efficient and less prone to garbage collection pauses than a Java-based storage engine. It is very fast for small datasets that fit in memory; datasets larger than memory are slower but still possible. It is [log-structured](http://www.igvita.com/2012/02/06/sstable-and-log-structured-storage-leveldb/), allowing very fast writes. It also includes support for block compression, which helps to reduce I/O and memory usage.
-Samza includes an additional in-memory caching layer in front of LevelDB, which avoids the cost of deserialization for frequently-accessed objects and batches writes. If the same key is updated multiple times in quick succession, the batching coalesces those updates into a single write. The writes are flushed to the changelog when a task [commits](checkpointing.html).
+Samza includes an additional in-memory caching layer in front of RocksDB, which avoids the cost of deserialization for frequently-accessed objects and batches writes. If the same key is updated multiple times in quick succession, the batching coalesces those updates into a single write. The writes are flushed to the changelog when a task [commits](checkpointing.html).
To use a key-value store in your job, add the following to your job config:
{% highlight jproperties %}
# Use the key-value store implementation for a store called "my-store"
-stores.my-store.factory=org.apache.samza.storage.kv.KeyValueStorageEngineFactory
+stores.my-store.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
# Use the Kafka topic "my-store-changelog" as the changelog stream for this store.
# This enables automatic recovery of the store after a failure. If you don't
@@ -182,11 +182,19 @@ public interface KeyValueStore<K, V> {
}
{% endhighlight %}
-Additional configuration properties for the key-value store are documented in the [configuration reference](../jobs/configuration-table.html#keyvalue).
+Additional configuration properties for the key-value store are documented in the [configuration reference](../jobs/configuration-table.html#keyvalue-rocksdb).
+
+#### Known Issues
+
+RocksDB has several rough edges. It's recommended that you read the RocksDB [tuning guide](https://github.com/facebook/rocksdb/wiki/RocksDB-Tuning-Guide). Some other notes to be aware of are:
+
+1. RocksDB is heavily optimized to run with SSD hard disks. Performance on non-SSDs degrades significantly.
+2. Samza's KeyValueStorageEngine.putAll() method does not currently use RocksDB's batching-put API because it's [non-functional in Java](https://github.com/facebook/rocksdb/issues/262).
+3. Calling iterator.seekToFirst() is very slow [if there are a lot of deletes in the store](https://github.com/facebook/rocksdb/issues/261).
### Implementing common use cases with the key-value store
-Earlier in this section we discussed some example use cases for stateful stream processing. Let's look at how each of these could be implemented using a key-value storage engine such as Samza's LevelDB.
+Earlier in this section we discussed some example use cases for stateful stream processing. Let's look at how each of these could be implemented using a key-value storage engine such as Samza's RocksDB store.
#### Windowed aggregation
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/0bab52d8/docs/learn/documentation/versioned/jobs/configuration-table.html
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index 3875145..e3251a6 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -844,13 +844,18 @@
<a href="../api/javadocs/org/apache/samza/task/InitableTask.html#init(org.apache.samza.config.Config, org.apache.samza.task.TaskContext)">init()</a>
method). The value of this property is the fully-qualified name of a Java class that implements
<a href="../api/javadocs/org/apache/samza/storage/StorageEngineFactory.html">StorageEngineFactory</a>.
- Samza currently ships with one storage engine implementation:
+ Samza currently ships with two storage engine implementations:
<dl>
- <dt><code>org.apache.samza.storage.kv.KeyValueStorageEngineFactory</code></dt>
+ <dt><code>org.apache.samza.storage.kv.LevelDbKeyValueStorageEngineFactory</code></dt>
<dd>An on-disk storage engine with a key-value interface, implemented using
<a href="https://code.google.com/p/leveldb/">LevelDB</a>. It supports fast random-access
reads and writes, as well as range queries on keys. LevelDB can be configured with
- various <a href="#keyvalue">additional tuning parameters</a>.</dd>
+ various <a href="#keyvalue-leveldb">additional tuning parameters</a>.</dd>
+ <dt><code>org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory</code></dt>
+ <dd>An on-disk storage engine with a key-value interface, implemented using
+ <a href="http://rocksdb.org/">RocksDB</a>. It supports fast random-access
+ reads and writes, as well as range queries on keys. RocksDB can be configured with
+ various <a href="#keyvalue-rocksdb">additional tuning parameters</a>.</dd>
</dl>
</td>
</tr>
@@ -900,18 +905,143 @@
</tr>
<tr>
- <th colspan="3" class="section" id="keyvalue">
+ <th colspan="3" class="section" id="keyvalue-rocksdb">
+ Using RocksDB for key-value storage<br>
+ <span class="subtitle">
+ (This section applies if you have set
+ <a href="#stores-factory" class="property">stores.*.factory</a>
+ <code>= org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory</code>)
+ </span>
+ </th>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-write-batch-size">stores.<span class="store">store-name</span>.<br>write.batch.size</td>
+ <td class="default">500</td>
+ <td class="description">
+ For better write performance, the storage engine buffers writes and applies them
+ to the underlying store in a batch. If the same key is written multiple times
+ in quick succession, this buffer also deduplicates writes to the same key. This
+ property is set to the number of key/value pairs that should be kept in this
+ in-memory buffer, per task instance. The number cannot be greater than
+ <a href="#stores-rocksdb-object-cache-size" class="property">stores.*.object.cache.size</a>.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-object-cache-size">stores.<span class="store">store-name</span>.<br>object.cache.size</td>
+ <td class="default">1000</td>
+ <td class="description">
+ Samza maintains an additional cache in front of RocksDB for frequently-accessed
+ objects. This cache contains deserialized objects (avoiding the deserialization
+ overhead on cache hits), in contrast to the RocksDB block cache
+ (<a href="#stores-rocksdb-container-cache-size-bytes" class="property">stores.*.container.cache.size.bytes</a>),
+ which caches serialized objects. This property determines the number of objects
+ to keep in Samza's cache, per task instance. This same cache is also used for
+ write buffering (see <a href="#stores-rocksdb-write-batch-size" class="property">stores.*.write.batch.size</a>).
+ A value of 0 disables all caching and batching.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-container-cache-size-bytes">stores.<span class="store">store-name</span>.container.<br>cache.size.bytes</td>
+ <td class="default">104857600</td>
+ <td class="description">
+ The size of RocksDB's block cache in bytes, per container. If there are several
+ task instances within one container, each is given a proportional share of this cache.
+ Note that this is an off-heap memory allocation, so the container's total memory use
+ is the maximum JVM heap size <em>plus</em> the size of this cache.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-container-write-buffer-size-bytes">stores.<span class="store">store-name</span>.container.<br>write.buffer.size.bytes</td>
+ <td class="default">33554432</td>
+ <td class="description">
+ The amount of memory (in bytes) that RocksDB uses for buffering writes before they are
+ written to disk, per container. If there are several task instances within one
+ container, each is given a proportional share of this buffer. This setting also
+ determines the size of RocksDB's segment files.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-compression">stores.<span class="store">store-name</span>.<br>rocksdb.compression</td>
+ <td class="default">snappy</td>
+ <td class="description">
+ This property controls whether RocksDB should compress data on disk and in the
+ block cache. The following values are valid:
+ <dl>
+ <dt><code>snappy</code></dt>
+ <dd>Compress data using the <a href="https://code.google.com/p/snappy/">Snappy</a> codec.</dd>
+ <dt><code>bzip2</code></dt>
+ <dd>Compress data using the <a href="http://en.wikipedia.org/wiki/Bzip2">bzip2</a> codec.</dd>
+ <dt><code>zlib</code></dt>
+ <dd>Compress data using the <a href="http://en.wikipedia.org/wiki/Zlib">zlib</a> codec.</dd>
+ <dt><code>lz4</code></dt>
+ <dd>Compress data using the <a href="https://code.google.com/p/lz4/">lz4</a> codec.</dd>
+ <dt><code>lz4hc</code></dt>
+ <dd>Compress data using the <a href="https://code.google.com/p/lz4/">lz4hc</a> (high compression) codec.</dd>
+ <dt><code>none</code></dt>
+ <dd>Do not compress data.</dd>
+ </dl>
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-block-size-bytes">stores.<span class="store">store-name</span>.<br>rocksdb.block.size.bytes</td>
+ <td class="default">4096</td>
+ <td class="description">
+ If compression is enabled, RocksDB groups approximately this many uncompressed bytes
+ into one compressed block. You probably don't need to change this property.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-bloomfilter-bits">stores.<span class="store">store-name</span>.<br>rocksdb.bloomfilter.bits</td>
+ <td class="default">10</td>
+ <td class="description">
+ In RocksDB, every SST file <a href="https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter">contains a Bloom filter</a>, which is used to determine if the file may contain a given key. Setting the bloom filter bit size allows developers to make the trade-off between the accuracy of the bloom filter, and its memory usage.
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-compaction-style">stores.<span class="store">store-name</span>.<br>rocksdb.compaction.style</td>
+ <td class="default">universal</td>
+ <td class="description">
+ This property controls the compaction style that RocksDB will employ when compacting its levels. The following values are valid:
+ <dl>
+ <dt><code>universal</code></dt>
+ <dd>Use <a href="https://github.com/facebook/rocksdb/wiki/Universal-Compaction">universal</a> compaction.</dd>
+ <dt><code>fifo</code></dt>
+ <dd>Use <a href="https://github.com/facebook/rocksdb/wiki/FIFO-compaction-style">FIFO</a> compaction.</dd>
+ <dt><code>level</code></dt>
+ <dd>Use LevelDB's standard leveled compaction.</dd>
+ </dl>
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="stores-rocksdb-num-write-buffers">stores.<span class="store">store-name</span>.<br>rocksdb.num.write.buffers</td>
+ <td class="default">3</td>
+ <td class="description">
+ Configures the <a href="https://github.com/facebook/rocksdb/wiki/Basic-Operations#write-buffer">number of write buffers</a> that a RocksDB store uses. This allows RocksDB to continue taking writes to other buffers even while a given write buffer is being flushed to disk.
+ </td>
+ </tr>
+
+ <tr>
+ <th colspan="3" class="section" id="keyvalue-leveldb">
Using LevelDB for key-value storage<br>
<span class="subtitle">
(This section applies if you have set
<a href="#stores-factory" class="property">stores.*.factory</a>
- <code>= org.apache.samza.storage.kv.KeyValueStorageEngineFactory</code>)
+ <code>= org.apache.samza.storage.kv.LevelDbKeyValueStorageEngineFactory</code>)
</span>
</th>
</tr>
<tr>
- <td class="property" id="stores-write-batch-size">stores.<span class="store">store-name</span>.<br>write.batch.size</td>
+ <td class="property" id="stores-leveldb-write-batch-size">stores.<span class="store">store-name</span>.<br>write.batch.size</td>
<td class="default">500</td>
<td class="description">
For better write performance, the storage engine buffers writes and applies them
@@ -919,27 +1049,27 @@
in quick succession, this buffer also deduplicates writes to the same key. This
property is set to the number of key/value pairs that should be kept in this
in-memory buffer, per task instance. The number cannot be greater than
- <a href="#stores-object-cache-size" class="property">stores.*.object.cache.size</a>.
+ <a href="#stores-leveldb-object-cache-size" class="property">stores.*.object.cache.size</a>.
</td>
</tr>
<tr>
- <td class="property" id="stores-object-cache-size">stores.<span class="store">store-name</span>.<br>object.cache.size</td>
+ <td class="property" id="stores-leveldb-object-cache-size">stores.<span class="store">store-name</span>.<br>object.cache.size</td>
<td class="default">1000</td>
<td class="description">
Samza maintains an additional cache in front of LevelDB for frequently-accessed
objects. This cache contains deserialized objects (avoiding the deserialization
overhead on cache hits), in contrast to the LevelDB block cache
- (<a href="#stores-container-cache-size-bytes" class="property">stores.*.container.cache.size.bytes</a>),
+ (<a href="#stores-leveldb-container-cache-size-bytes" class="property">stores.*.container.cache.size.bytes</a>),
which caches serialized objects. This property determines the number of objects
to keep in Samza's cache, per task instance. This same cache is also used for
- write buffering (see <a href="#stores-write-batch-size" class="property">stores.*.write.batch.size</a>).
+ write buffering (see <a href="#stores-leveldb-write-batch-size" class="property">stores.*.write.batch.size</a>).
A value of 0 disables all caching and batching.
</td>
</tr>
<tr>
- <td class="property" id="stores-container-cache-size-bytes">stores.<span class="store">store-name</span>.container.<br>cache.size.bytes</td>
+ <td class="property" id="stores-leveldb-container-cache-size-bytes">stores.<span class="store">store-name</span>.container.<br>cache.size.bytes</td>
<td class="default">104857600</td>
<td class="description">
The size of LevelDB's block cache in bytes, per container. If there are several
@@ -950,7 +1080,7 @@
</tr>
<tr>
- <td class="property" id="stores-container-write-buffer-size-bytes">stores.<span class="store">store-name</span>.container.<br>write.buffer.size.bytes</td>
+ <td class="property" id="stores-leveldb-container-write-buffer-size-bytes">stores.<span class="store">store-name</span>.container.<br>write.buffer.size.bytes</td>
<td class="default">33554432</td>
<td class="description">
The amount of memory (in bytes) that LevelDB uses for buffering writes before they are
@@ -961,7 +1091,7 @@
</tr>
<tr>
- <td class="property" id="stores-compaction-delete-threshold">stores.<span class="store">store-name</span>.<br>compaction.delete.threshold</td>
+ <td class="property" id="stores-leveldb-compaction-delete-threshold">stores.<span class="store">store-name</span>.<br>compaction.delete.threshold</td>
<td class="default">-1</td>
<td class="description">
Setting this property forces a LevelDB compaction to be performed after a certain
@@ -1041,7 +1171,7 @@
memory use remains below the limit. The amount of memory used is normally the JVM heap
size (configured with <a href="#task-opts" class="property">task.opts</a>), plus the
size of any off-heap memory allocation (for example
- <a href="#stores-container-cache-size-bytes" class="property">stores.*.container.cache.size.bytes</a>),
+ <a href="#stores-rocksdb-container-cache-size-bytes" class="property">stores.*.container.cache.size.bytes</a>),
plus a safety margin to allow for JVM overheads.
</td>
</tr>
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/0bab52d8/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
index a52731b..14eeba5 100644
--- a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
+++ b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.samza.storage
+package org.apache.samza.storage.kv
import java.io.File
import org.apache.samza.container.SamzaContainerContext