You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/02/11 22:01:21 UTC
[2/3] flink git commit: [FLINK-3389] [rocksdb] Add pre-defined option
profiles.
[FLINK-3389] [rocksdb] Add pre-defined option profiles.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/be72758d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/be72758d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/be72758d
Branch: refs/heads/master
Commit: be72758d1104400c8a48554d717c5b8cea5b3617
Parents: 82c7383
Author: Stephan Ewen <se...@apache.org>
Authored: Thu Feb 11 15:30:56 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Feb 11 21:34:03 2016 +0100
----------------------------------------------------------------------
.../contrib/streaming/state/OptionsFactory.java | 32 ++++++-
.../streaming/state/PredefinedOptions.java | 91 ++++++++++++++++++++
.../streaming/state/RocksDBStateBackend.java | 76 ++++++++++++++--
3 files changed, 190 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/be72758d/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
index 73b1e5d..3e52f1f 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
@@ -24,8 +24,36 @@ import org.rocksdb.Options;
* A factory for {@link Options} to be passed to the {@link RocksDBStateBackend}.
* Options have to be created lazily by this factory, because the {@code Options}
* class is not serializable and holds pointers to native code.
+ *
+ * <p>A typical pattern to use this OptionsFactory is as follows:
+ *
+ * <h3>Java 8:</h3>
+ * <pre>{@code
+ * rocksDbBackend.setOptions( (currentOptions) -> currentOptions.setMaxOpenFiles(1024) );
+ * }</pre>
+ *
+ * <h3>Java 7:</h3>
+ * <pre>{@code
+ * rocksDbBackend.setOptions(new OptionsFactory() {
+ *
+ * public Options setOptions(Options currentOptions) {
+ * return currentOptions.setMaxOpenFiles(1024);
+ * }
+ * })
+ * }</pre>
*/
public interface OptionsFactory extends java.io.Serializable {
-
- Options createOptions();
+
+ /**
+ * This method should set the additional options on top of the current options object.
+ * The current options object may contain pre-defined options based on flags that have
+ * been configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from
+ * the setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @return The options object on which the additional options are set.
+ */
+ Options createOptions(Options currentOptions);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/be72758d/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/PredefinedOptions.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/PredefinedOptions.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/PredefinedOptions.java
new file mode 100644
index 0000000..383f043
--- /dev/null
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/PredefinedOptions.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.contrib.streaming.state;
+
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.Options;
+
+/**
+ * The {@code PredefinedOptions} are configuration settings for the {@link RocksDBStateBackend}.
+ * The various pre-defined choices are configurations that have been empirically
+ * determined to be beneficial for performance under different settings.
+ *
+ * <p>Some of these settings are based on experiments by the Flink community, some follow
+ * guides from the RocksDB project.
+ */
+public enum PredefinedOptions {
+
+ /**
+ * Default options for all settings.
+ */
+ DEFAULT {
+
+ @Override
+ public Options createOptions() {
+ return new Options();
+ }
+ },
+
+ /**
+ * Pre-defined options for regular spinning hard disks.
+ *
+ * <p>This constant configures RocksDB with some options that lead empirically
+ * to better performance when the machines executing the system use
+ * regular spinning hard disks. The following options are set:
+ * <ul>
+ * <li>Optimized level-style compactions</li>
+ * </ul>
+ */
+ SPINNING_DISK_OPTIMIZED {
+
+ @Override
+ public Options createOptions() {
+ return new Options()
+ .setCompactionStyle(CompactionStyle.LEVEL)
+ .optimizeLevelStyleCompaction();
+ }
+ },
+
+ /**
+ * Pre-defined options for Flash SSDs.
+ *
+ * <p>This constant configures RocksDB with some options that lead empirically
+ * to better performance when the machines executing the system use SSDs.
+ * The following options are set:
+ * <ul>
+ * <li>none</li>
+ * </ul>
+ */
+ FLASH_SSD_OPTIMIZED {
+
+ @Override
+ public Options createOptions() {
+ return new Options();
+ }
+ };
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates the options for this pre-defined setting.
+ *
+ * @return The pre-defined options object.
+ */
+ public abstract Options createOptions();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/be72758d/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index 8c0171a..eddd8c0 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -32,6 +32,7 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.api.common.state.StateBackend;
+
import org.rocksdb.Options;
import org.rocksdb.StringAppendOperator;
@@ -45,6 +46,10 @@ import static java.util.Objects.requireNonNull;
* For persistence against loss of machines, checkpoints take a snapshot of the
* RocksDB database, and persist that snapshot in a file system (by default) or
* another configurable state backend.
+ *
+ * <p>The behavior of the RocksDB instances can be parametrized by setting RocksDB Options
+ * using the methods {@link #setPredefinedOptions(PredefinedOptions)} and
+ * {@link #setOptions(OptionsFactory)}.
*/
public class RocksDBStateBackend extends AbstractStateBackend {
private static final long serialVersionUID = 1L;
@@ -62,6 +67,9 @@ public class RocksDBStateBackend extends AbstractStateBackend {
private JobID jobId;
private AbstractStateBackend backingStateBackend;
+
+ /** The pre-configured option settings */
+ private PredefinedOptions predefinedOptions = PredefinedOptions.DEFAULT;
/** The options factory to create the RocksDB options in the cluster */
private OptionsFactory optionsFactory;
@@ -69,6 +77,7 @@ public class RocksDBStateBackend extends AbstractStateBackend {
/** The options from the options factory, cached */
private transient Options rocksDbOptions;
+ // ------------------------------------------------------------------------
public RocksDBStateBackend(String dbBasePath, String checkpointDirectory, AbstractStateBackend backingStateBackend) {
this.dbBasePath = requireNonNull(dbBasePath);
@@ -76,10 +85,14 @@ public class RocksDBStateBackend extends AbstractStateBackend {
this.backingStateBackend = requireNonNull(backingStateBackend);
}
+ // ------------------------------------------------------------------------
+
@Override
- public void initializeForJob(Environment env,
- String operatorIdentifier,
- TypeSerializer<?> keySerializer) throws Exception {
+ public void initializeForJob(
+ Environment env,
+ String operatorIdentifier,
+ TypeSerializer<?> keySerializer) throws Exception
+ {
super.initializeForJob(env, operatorIdentifier, keySerializer);
this.operatorIdentifier = operatorIdentifier.replace(" ", "");
backingStateBackend.initializeForJob(env, operatorIdentifier, keySerializer);
@@ -106,6 +119,10 @@ public class RocksDBStateBackend extends AbstractStateBackend {
return checkpointDirectory + "/" + jobId.toString() + "/" + operatorIdentifier + "/" + stateName;
}
+ // ------------------------------------------------------------------------
+ // State factories
+ // ------------------------------------------------------------------------
+
@Override
protected <N, T> ValueState<T> createValueState(TypeSerializer<N> namespaceSerializer,
ValueStateDescriptor<T> stateDesc) throws Exception {
@@ -154,9 +171,42 @@ public class RocksDBStateBackend extends AbstractStateBackend {
// ------------------------------------------------------------------------
/**
- * Defines the {@link org.rocksdb.Options} for the RocksDB instances.
+ * Sets the predefined options for RocksDB.
+ *
+ * <p>If a user-defined options factory is set (via {@link #setOptions(OptionsFactory)}),
+ * then the options from the factory are applied on top of the here specified
+ * predefined options.
+ *
+ * @param options The options to set (must not be null).
+ */
+ public void setPredefinedOptions(PredefinedOptions options) {
+ predefinedOptions = requireNonNull(options);
+ }
+
+ /**
+ * Gets the currently set predefined options for RocksDB.
+ * The default options (if nothing was set via {@link #setPredefinedOptions(PredefinedOptions)})
+ * are {@link PredefinedOptions#DEFAULT}.
+ *
+ * <p>If a user-defined options factory is set (via {@link #setOptions(OptionsFactory)}),
+ * then the options from the factory are applied on top of the predefined options.
+ *
+ * @return The currently set predefined options for RocksDB.
+ */
+ public PredefinedOptions getPredefinedOptions() {
+ return predefinedOptions;
+ }
+
+ /**
+ * Sets {@link org.rocksdb.Options} for the RocksDB instances.
* Because the options are not serializable and hold native code references,
- * they must be specified through a factory.
+ * they must be specified through a factory.
+ *
+ * <p>The options created by the factory here are applied on top of the pre-defined
+ * options profile selected via {@link #setPredefinedOptions(PredefinedOptions)}.
+ * If the pre-defined options profile is the default
+ * ({@link PredefinedOptions#DEFAULT}), then the factory fully controls the RocksDB
+ * options.
*
* @param optionsFactory The options factory that lazily creates the RocksDB options.
*/
@@ -172,12 +222,24 @@ public class RocksDBStateBackend extends AbstractStateBackend {
public OptionsFactory getOptions() {
return optionsFactory;
}
-
+
+ /**
+ * Gets the RocksDB Options to be used for all RocksDB instances.
+ */
Options getRocksDBOptions() {
if (rocksDbOptions == null) {
- Options opt = optionsFactory == null ? new Options() : optionsFactory.createOptions();
+ // initial options from pre-defined profile
+ Options opt = predefinedOptions.createOptions();
+
+ // add user-defined options, if specified
+ if (optionsFactory != null) {
+ opt = optionsFactory.createOptions(opt);
+ }
+
+ // add necessary default options
opt = opt.setCreateIfMissing(true);
opt = opt.setMergeOperator(new StringAppendOperator());
+
rocksDbOptions = opt;
}
return rocksDbOptions;