From e6db3711eca8f9e7321dfefcecee343d747db671 Mon Sep 17 00:00:00 2001 From: Lim Chee Hau Date: Sat, 24 Jun 2017 16:26:15 +0200 Subject: [PATCH] [FLINK-7000] Add custom configuration local environment in Scala StreamExecutionEnvironment --- .../api/scala/StreamExecutionEnvironment.scala | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala index 742baf9315452..9fd03c3d8c290 100644 --- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala +++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala @@ -767,6 +767,18 @@ object StreamExecutionEnvironment { new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism)) } + /** + * Creates a local execution environment. The local execution environment will run the + * program in a multi-threaded fashion in the same JVM as the environment was created in. + * + * @param parallelism The parallelism for the local environment. + * @param configuration Pass a custom configuration into the cluster. + */ + def createLocalEnvironment(parallelism: Int, configuration: Configuration): + StreamExecutionEnvironment = { + new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, configuration)) + } + /** * Creates a [[StreamExecutionEnvironment]] for local program execution that also starts the * web monitoring UI.