You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by za...@apache.org on 2019/10/03 08:38:48 UTC
[ignite] branch master updated: IGNITE-11723:[Spark] IgniteSpark
integration should support skipStore option for internal dataStreamer
(IgniteRdd and Ignite DataFrame) (#6907)
This is an automated email from the ASF dual-hosted git repository.
zaleslaw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 2e7553a IGNITE-11723:[Spark] IgniteSpark integration should support skipStore option for internal dataStreamer (IgniteRdd and Ignite DataFrame) (#6907)
2e7553a is described below
commit 2e7553aa469a679c8a297e49f50b5464d9d76488
Author: Alexey Zinoviev <za...@gmail.com>
AuthorDate: Thu Oct 3 11:38:33 2019 +0300
IGNITE-11723:[Spark] IgniteSpark integration should support skipStore option for internal dataStreamer (IgniteRdd and Ignite DataFrame) (#6907)
---
.../ignite/spark/IgniteDataFrameSettings.scala | 19 ++
.../scala/org/apache/ignite/spark/IgniteRDD.scala | 11 +-
.../org/apache/ignite/spark/JavaIgniteRDD.scala | 13 +-
.../ignite/spark/impl/IgniteRelationProvider.scala | 3 +
.../org/apache/ignite/spark/impl/QueryHelper.scala | 9 +-
.../spark/JavaEmbeddedIgniteRDDSelfTest.java | 6 +-
...avaEmbeddedIgniteRDDWithLocalStoreSelfTest.java | 227 +++++++++++++++++++++
.../ignite/testsuites/IgniteRDDTestSuite.java | 4 +-
8 files changed, 276 insertions(+), 16 deletions(-)
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteDataFrameSettings.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteDataFrameSettings.scala
index e176721..4e0abf4 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteDataFrameSettings.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteDataFrameSettings.scala
@@ -120,6 +120,25 @@ object IgniteDataFrameSettings {
/**
* Config option for saving data frame.
* Internally all SQL inserts are done through `IgniteDataStreamer`.
+ * This options sets `skipStore` property of streamer.
+ * If `true` then write-through behavior will be disabled for data streaming.
+ * If `false` then write-through behavior will be enabled for data streaming.
+ * Default value if `false`.
+ *
+ * @example {{{
+ * val igniteDF = spark.write.format(IGNITE)
+ * // other options ...
+ * .option(OPTION_STREAMER_SKIP_STORE, true)
+ * .save()
+ * }}}
+ * @see [[org.apache.ignite.IgniteDataStreamer]]
+ * @see [[org.apache.ignite.IgniteDataStreamer#skipStore(boolean)]]
+ */
+ val OPTION_STREAMER_SKIP_STORE = "streamerSkipStore"
+
+ /**
+ * Config option for saving data frame.
+ * Internally all SQL inserts are done through `IgniteDataStreamer`.
* This options sets `autoFlushFrequency` property of streamer.
*
* @example {{{
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteRDD.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteRDD.scala
index 5fb81b6..25784d1 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteRDD.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteRDD.scala
@@ -17,7 +17,6 @@
package org.apache.ignite.spark
import javax.cache.Cache
-
import org.apache.ignite.cache.query._
import org.apache.ignite.cluster.ClusterNode
import org.apache.ignite.configuration.CacheConfiguration
@@ -228,8 +227,9 @@ class IgniteRDD[K, V] (
* @param rdd RDD instance to save values from.
* @param overwrite Boolean flag indicating whether the call on this method should overwrite existing
* values in Ignite cache.
+ * @param skipStore Sets flag indicating that write-through behavior should be disabled for data streaming.
*/
- def savePairs(rdd: RDD[(K, V)], overwrite: Boolean = false) = {
+ def savePairs(rdd: RDD[(K, V)], overwrite: Boolean = false, skipStore: Boolean = false) = {
rdd.foreachPartition(it ⇒ {
val ig = ic.ignite()
@@ -240,6 +240,7 @@ class IgniteRDD[K, V] (
try {
streamer.allowOverwrite(overwrite)
+ streamer.skipStore(skipStore)
it.foreach(tup ⇒ {
streamer.addData(tup._1, tup._2)
@@ -258,8 +259,9 @@ class IgniteRDD[K, V] (
* @param f Transformation function.
* @param overwrite Boolean flag indicating whether the call on this method should overwrite existing
* values in Ignite cache.
+ * @param skipStore Sets flag indicating that write-through behavior should be disabled for data streaming.
*/
- def savePairs[T](rdd: RDD[T], f: (T, IgniteContext) ⇒ (K, V), overwrite: Boolean) = {
+ def savePairs[T](rdd: RDD[T], f: (T, IgniteContext) ⇒ (K, V), overwrite: Boolean, skipStore: Boolean) = {
rdd.foreachPartition(it ⇒ {
val ig = ic.ignite()
@@ -270,6 +272,7 @@ class IgniteRDD[K, V] (
try {
streamer.allowOverwrite(overwrite)
+ streamer.skipStore(skipStore)
it.foreach(t ⇒ {
val tup = f(t, ic)
@@ -290,7 +293,7 @@ class IgniteRDD[K, V] (
* @param f Transformation function.
*/
def savePairs[T](rdd: RDD[T], f: (T, IgniteContext) ⇒ (K, V)): Unit = {
- savePairs(rdd, f, overwrite = false)
+ savePairs(rdd, f, overwrite = false, skipStore = false)
}
/**
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/JavaIgniteRDD.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/JavaIgniteRDD.scala
index a44cb51..1937483 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/JavaIgniteRDD.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/JavaIgniteRDD.scala
@@ -82,20 +82,21 @@ class JavaIgniteRDD[K, V](override val rdd: IgniteRDD[K, V])
def saveValues[T](jrdd: JavaRDD[T], f: (T, IgniteContext) ⇒ V) = rdd.saveValues(JavaRDD.toRDD(jrdd), f)
- def savePairs(jrdd: JavaPairRDD[K, V], overwrite: Boolean) = {
+ def savePairs(jrdd: JavaPairRDD[K, V], overwrite: Boolean, skipStore: Boolean) = {
val rrdd: RDD[(K, V)] = JavaPairRDD.toRDD(jrdd)
- rdd.savePairs(rrdd, overwrite)
+ rdd.savePairs(rrdd, overwrite, skipStore)
}
- def savePairs(jrdd: JavaPairRDD[K, V]) : Unit = savePairs(jrdd, overwrite = false)
+ def savePairs(jrdd: JavaPairRDD[K, V]): Unit = savePairs(jrdd, overwrite = false, skipStore = false)
- def savePairs[T](jrdd: JavaRDD[T], f: (T, IgniteContext) ⇒ (K, V), overwrite: Boolean = false) = {
- rdd.savePairs(JavaRDD.toRDD(jrdd), f, overwrite)
+ def savePairs[T](jrdd: JavaRDD[T], f: (T, IgniteContext) ⇒ (K, V), overwrite: Boolean = false,
+ skipStore: Boolean = false) = {
+ rdd.savePairs(JavaRDD.toRDD(jrdd), f, overwrite, skipStore)
}
def savePairs[T](jrdd: JavaRDD[T], f: (T, IgniteContext) ⇒ (K, V)): Unit =
- savePairs(jrdd, f, overwrite = false)
+ savePairs(jrdd, f, overwrite = false, skipStore = false)
def clear(): Unit = rdd.clear()
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/impl/IgniteRelationProvider.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/impl/IgniteRelationProvider.scala
index 039ca63..a4f6da1 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/impl/IgniteRelationProvider.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/impl/IgniteRelationProvider.scala
@@ -125,6 +125,7 @@ class IgniteRelationProvider extends RelationProvider
params.get(OPTION_SCHEMA),
ctx,
params.get(OPTION_STREAMER_ALLOW_OVERWRITE).map(_.toBoolean),
+ params.get(OPTION_STREAMER_SKIP_STORE).map(_.toBoolean),
params.get(OPTION_STREAMER_FLUSH_FREQUENCY).map(_.toLong),
params.get(OPTION_STREAMER_PER_NODE_BUFFER_SIZE).map(_.toInt),
params.get(OPTION_STREAMER_PER_NODE_PARALLEL_OPERATIONS).map(_.toInt))
@@ -135,6 +136,7 @@ class IgniteRelationProvider extends RelationProvider
params.get(OPTION_SCHEMA),
ctx,
params.get(OPTION_STREAMER_ALLOW_OVERWRITE).map(_.toBoolean),
+ params.get(OPTION_STREAMER_SKIP_STORE).map(_.toBoolean),
params.get(OPTION_STREAMER_FLUSH_FREQUENCY).map(_.toLong),
params.get(OPTION_STREAMER_PER_NODE_BUFFER_SIZE).map(_.toInt),
params.get(OPTION_STREAMER_PER_NODE_PARALLEL_OPERATIONS).map(_.toInt))
@@ -166,6 +168,7 @@ class IgniteRelationProvider extends RelationProvider
params.get(OPTION_SCHEMA),
ctx,
params.get(OPTION_STREAMER_ALLOW_OVERWRITE).map(_.toBoolean),
+ params.get(OPTION_STREAMER_SKIP_STORE).map(_.toBoolean),
params.get(OPTION_STREAMER_FLUSH_FREQUENCY).map(_.toLong),
params.get(OPTION_STREAMER_PER_NODE_BUFFER_SIZE).map(_.toInt),
params.get(OPTION_STREAMER_PER_NODE_PARALLEL_OPERATIONS).map(_.toInt))
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/impl/QueryHelper.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/impl/QueryHelper.scala
index f752b1a..d123b01 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/impl/QueryHelper.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/impl/QueryHelper.scala
@@ -18,12 +18,12 @@
package org.apache.ignite.spark.impl
import org.apache.ignite.cache.query.SqlFieldsQuery
-import org.apache.ignite.spark.IgniteDataFrameSettings._
-import QueryUtils.{compileCreateTable, compileDropTable, compileInsert}
import org.apache.ignite.internal.IgniteEx
import org.apache.ignite.internal.processors.query.QueryTypeDescriptorImpl
import org.apache.ignite.internal.processors.query.QueryUtils.DFLT_SCHEMA
import org.apache.ignite.spark.IgniteContext
+import org.apache.ignite.spark.IgniteDataFrameSettings._
+import org.apache.ignite.spark.impl.QueryUtils.{compileCreateTable, compileDropTable, compileInsert}
import org.apache.ignite.{Ignite, IgniteException}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, Row}
@@ -116,6 +116,7 @@ private[apache] object QueryHelper {
schemaName: Option[String],
ctx: IgniteContext,
streamerAllowOverwrite: Option[Boolean],
+ streamerSkipStore: Option[Boolean],
streamerFlushFrequency: Option[Long],
streamerPerNodeBufferSize: Option[Int],
streamerPerNodeParallelOperations: Option[Int]
@@ -129,6 +130,7 @@ private[apache] object QueryHelper {
schemaName,
ctx,
streamerAllowOverwrite,
+ streamerSkipStore,
streamerFlushFrequency,
streamerPerNodeBufferSize,
streamerPerNodeParallelOperations
@@ -160,6 +162,7 @@ private[apache] object QueryHelper {
schemaName: Option[String],
ctx: IgniteContext,
streamerAllowOverwrite: Option[Boolean],
+ streamerSkipStore: Option[Boolean],
streamerFlushFrequency: Option[Long],
streamerPerNodeBufferSize: Option[Int],
streamerPerNodeParallelOperations: Option[Int]
@@ -170,6 +173,8 @@ private[apache] object QueryHelper {
streamerAllowOverwrite.foreach(v ⇒ streamer.allowOverwrite(v))
+ streamerSkipStore.foreach(v ⇒ streamer.skipStore(v))
+
streamerFlushFrequency.foreach(v ⇒ streamer.autoFlushFrequency(v))
streamerPerNodeBufferSize.foreach(v ⇒ streamer.perNodeBufferSize(v))
diff --git a/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDSelfTest.java b/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDSelfTest.java
index bf256c6..fbd4363 100644
--- a/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDSelfTest.java
+++ b/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDSelfTest.java
@@ -129,7 +129,7 @@ public class JavaEmbeddedIgniteRDDSelfTest extends GridCommonAbstractTest {
ic = new JavaIgniteContext<>(sc, new IgniteConfigProvider(), false);
ic.fromCache(PARTITIONED_CACHE_NAME)
- .savePairs(sc.parallelize(F.range(0, KEYS_CNT), GRID_CNT).mapToPair(TO_PAIR_F), true);
+ .savePairs(sc.parallelize(F.range(0, KEYS_CNT), GRID_CNT).mapToPair(TO_PAIR_F), true, false);
Ignite ignite = ic.ignite();
@@ -200,7 +200,7 @@ public class JavaEmbeddedIgniteRDDSelfTest extends GridCommonAbstractTest {
JavaIgniteRDD<String, Entity> cache = ic.fromCache(PARTITIONED_CACHE_NAME);
int cnt = 1001;
- cache.savePairs(sc.parallelize(F.range(0, cnt), GRID_CNT).mapToPair(INT_TO_ENTITY_F), true);
+ cache.savePairs(sc.parallelize(F.range(0, cnt), GRID_CNT).mapToPair(INT_TO_ENTITY_F), true, false);
List<Entity> res = cache.objectSql("Entity", "name = ? and salary = ?", "name50", 5000)
.map(STR_ENTITY_PAIR_TO_ENTITY_F).collect();
@@ -238,7 +238,7 @@ public class JavaEmbeddedIgniteRDDSelfTest extends GridCommonAbstractTest {
JavaIgniteRDD<String, Entity> cache = ic.fromCache(PARTITIONED_CACHE_NAME);
- cache.savePairs(sc.parallelize(F.range(0, 1001), GRID_CNT).mapToPair(INT_TO_ENTITY_F), true);
+ cache.savePairs(sc.parallelize(F.range(0, 1001), GRID_CNT).mapToPair(INT_TO_ENTITY_F), true, false);
Dataset<Row> df =
cache.sql("select id, name, salary from Entity where name = ? and salary = ?", "name50", 5000);
diff --git a/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDWithLocalStoreSelfTest.java b/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDWithLocalStoreSelfTest.java
new file mode 100644
index 0000000..e5c1f4b
--- /dev/null
+++ b/modules/spark/src/test/java/org/apache/ignite/spark/JavaEmbeddedIgniteRDDWithLocalStoreSelfTest.java
@@ -0,0 +1,227 @@
+/*
+ * 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.ignite.spark;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.Cache;
+import javax.cache.configuration.FactoryBuilder;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.store.CacheStoreAdapter;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgniteOutClosure;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.Function2;
+import org.apache.spark.api.java.function.PairFunction;
+import org.apache.spark.sql.Column;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+import scala.Tuple2;
+
+/**
+ * Tests for {@link JavaIgniteRDD} (embedded mode).
+ */
+public class JavaEmbeddedIgniteRDDWithLocalStoreSelfTest extends GridCommonAbstractTest {
+ /** */
+ private static ConcurrentHashMap<Object, Object> storeMap;
+
+ /** */
+ private TestStore store;
+
+ /** For Ignite instance names generation */
+ private static AtomicInteger cntr = new AtomicInteger(1);
+
+ /** Ignite instance names. */
+ private static ThreadLocal<Integer> igniteInstanceNames = new ThreadLocal<Integer>() {
+ @Override protected Integer initialValue() {
+ return cntr.getAndIncrement();
+ }
+ };
+
+ /** Grid count. */
+ private static final int GRID_CNT = 3;
+
+ /** Cache name. */
+ private static final String PARTITIONED_CACHE_NAME = "partitioned";
+
+ /** To pair function. */
+ private static final PairFunction<Integer, Integer, Integer> SIMPLE_FUNCTION = new PairFunction<Integer, Integer, Integer>() {
+ /** {@inheritDoc} */
+ @Override public Tuple2<Integer, Integer> call(Integer i) {
+ return new Tuple2<>(i, i);
+ }
+ };
+
+ /**
+ * Default constructor.
+ */
+ public JavaEmbeddedIgniteRDDWithLocalStoreSelfTest() {
+ super(false);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * Creates default spark context
+ *
+ * @return Context.
+ */
+ private JavaSparkContext createContext() {
+ SparkConf conf = new SparkConf();
+
+ conf.set("spark.executor.instances", String.valueOf(GRID_CNT));
+
+ return new JavaSparkContext("local[" + GRID_CNT + "]", "test", conf);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testStoreDataToIgniteWithOptionSkipStore() throws Exception {
+ storeMap = new ConcurrentHashMap<>();
+ store = new TestStore();
+
+ JavaSparkContext sc = createContext();
+
+ JavaIgniteContext<Integer, Integer> ic = null;
+
+ try {
+ ic = new JavaIgniteContext<>(sc, new IgniteConfigProvider(), false);
+
+ Ignite ignite = ic.ignite();
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(PARTITIONED_CACHE_NAME);
+
+ for (int i = 0; i < 1000; i++)
+ storeMap.put(i, i);
+
+ ic.fromCache(PARTITIONED_CACHE_NAME)
+ .savePairs(sc.parallelize(F.range(1000, 2000), GRID_CNT).mapToPair(SIMPLE_FUNCTION), true, false);
+
+ for (int i = 0; i < 2000; i++)
+ assertEquals(i, storeMap.get(i));
+
+ ic.fromCache(PARTITIONED_CACHE_NAME)
+ .savePairs(sc.parallelize(F.range(2000, 3000), GRID_CNT).mapToPair(SIMPLE_FUNCTION), true, true);
+
+ for (int i = 2000; i < 3000; i++)
+ assertNull(storeMap.get(i));
+
+ for (int i = 0; i < 3000; i++) {
+ Integer val = cache.get(i);
+
+ assertNotNull("Value was not put to cache for key: " + i, val);
+ assertEquals("Invalid value stored for key: " + i, Integer.valueOf(i), val);
+ }
+ }
+ finally {
+ if (ic != null)
+ ic.close(true);
+
+ sc.stop();
+ }
+ }
+
+ /**
+ * @param igniteInstanceName Ignite instance name.
+ * @param client Client.
+ * @throws Exception If failed.
+ * @return Confiuration.
+ */
+ private static IgniteConfiguration getConfiguration(String igniteInstanceName, boolean client) throws Exception {
+ IgniteConfiguration cfg = new IgniteConfiguration();
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+
+ cfg.setClientMode(client);
+
+ cfg.setIgniteInstanceName(igniteInstanceName);
+
+ return cfg;
+ }
+
+ /**
+ * Creates cache configuration.
+ *
+ * @return Cache configuration.
+ */
+ private static CacheConfiguration<Object, Object> cacheConfiguration() {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
+
+ ccfg.setBackups(1);
+
+ ccfg.setName(PARTITIONED_CACHE_NAME);
+
+ ccfg.setIndexedTypes(String.class, Entity.class);
+
+ ccfg.setCacheStoreFactory(FactoryBuilder.factoryOf(TestStore.class));
+
+ ccfg.setReadThrough(true);
+ ccfg.setWriteThrough(true);
+
+ return ccfg;
+ }
+
+ /**
+ * Ignite configiration provider.
+ */
+ static class IgniteConfigProvider implements IgniteOutClosure<IgniteConfiguration> {
+ /** {@inheritDoc} */
+ @Override public IgniteConfiguration apply() {
+ try {
+ return getConfiguration("worker-" + igniteInstanceNames.get(), false);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /** */
+ @SuppressWarnings("PublicInnerClass")
+ public static class TestStore extends CacheStoreAdapter<Object, Object> {
+ /** {@inheritDoc} */
+ @Nullable @Override public Object load(Object key) {
+ return storeMap.get(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(Cache.Entry<?, ?> entry) {
+ storeMap.put(entry.getKey(), entry.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void delete(Object key) {
+ storeMap.remove(key);
+ }
+ }
+}
diff --git a/modules/spark/src/test/java/org/apache/ignite/testsuites/IgniteRDDTestSuite.java b/modules/spark/src/test/java/org/apache/ignite/testsuites/IgniteRDDTestSuite.java
index b1ec515..bde086d 100644
--- a/modules/spark/src/test/java/org/apache/ignite/testsuites/IgniteRDDTestSuite.java
+++ b/modules/spark/src/test/java/org/apache/ignite/testsuites/IgniteRDDTestSuite.java
@@ -18,6 +18,7 @@
package org.apache.ignite.testsuites;
import org.apache.ignite.spark.JavaEmbeddedIgniteRDDSelfTest;
+import org.apache.ignite.spark.JavaEmbeddedIgniteRDDWithLocalStoreSelfTest;
import org.apache.ignite.spark.JavaStandaloneIgniteRDDSelfTest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
@@ -28,7 +29,8 @@ import org.junit.runners.Suite;
@RunWith(Suite.class)
@Suite.SuiteClasses({
JavaEmbeddedIgniteRDDSelfTest.class,
- JavaStandaloneIgniteRDDSelfTest.class
+ JavaStandaloneIgniteRDDSelfTest.class,
+ JavaEmbeddedIgniteRDDWithLocalStoreSelfTest.class
})
public class IgniteRDDTestSuite {
}