ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [06/50] incubator-ignite git commit: #IGNITE-389 - Adding test, fixing colocation.
Date Thu, 11 Jun 2015 07:18:21 GMT
#IGNITE-389 - Adding test, fixing colocation.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/f00a9e99
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/f00a9e99
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/f00a9e99

Branch: refs/heads/ignite-484-1
Commit: f00a9e9980aefeba2d80969730552e5c5651f1c6
Parents: d151244
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Wed May 27 23:30:49 2015 -0700
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Wed May 27 23:30:49 2015 -0700

----------------------------------------------------------------------
 modules/spark/pom.xml                           |  8 ----
 .../org/apache/ignite/spark/IgniteRDD.scala     |  8 ++--
 .../ignite/spark/examples/ColocationTest.scala  | 40 ++++++++++++++++++++
 3 files changed, 45 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f00a9e99/modules/spark/pom.xml
----------------------------------------------------------------------
diff --git a/modules/spark/pom.xml b/modules/spark/pom.xml
index dc01c76..84055d6 100644
--- a/modules/spark/pom.xml
+++ b/modules/spark/pom.xml
@@ -49,14 +49,6 @@
         </dependency>
 
         <dependency>
-            <groupId>org.apache.ignite</groupId>
-            <artifactId>ignite-indexing</artifactId>
-            <version>${project.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
-
-        <dependency>
             <groupId>org.scala-lang</groupId>
             <artifactId>scala-library</artifactId>
             <version>2.10.4</version>

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f00a9e99/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteRDD.scala
----------------------------------------------------------------------
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 ee0e9b3..30efa7a 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
@@ -23,6 +23,7 @@ import org.apache.ignite.cluster.ClusterNode
 import org.apache.ignite.configuration.CacheConfiguration
 import org.apache.ignite.lang.IgniteUuid
 import org.apache.ignite.spark.impl.{IgniteAbstractRDD, IgniteSqlRDD, IgnitePartition, IgniteQueryIterator}
+import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode
 import org.apache.spark.rdd.RDD
 import org.apache.spark.{TaskContext, Partition}
 
@@ -37,9 +38,9 @@ class IgniteRDD[K, V] (
     override def compute(part: Partition, context: TaskContext): Iterator[(K, V)] = {
         val cache = ensureCache()
 
-        val qry: ScanQuery[K, V] = new ScanQuery[K, V]()
+        val qry: ScanQuery[K, V] = new ScanQuery[K, V](part.index)
 
-        qry.setPartition(part.index)
+        val partNodes = ic.ignite().affinity(cache.getName).mapPartitionToPrimaryAndBackups(part.index)
 
         val it: java.util.Iterator[Cache.Entry[K, V]] = cache.query(qry).iterator()
 
@@ -59,7 +60,8 @@ class IgniteRDD[K, V] (
     override protected def getPreferredLocations(split: Partition): Seq[String] = {
         ensureCache()
 
-        ic.ignite().affinity(cacheName).mapPartitionToPrimaryAndBackups(split.index).map(_.addresses()).flatten.toList
+        ic.ignite().affinity(cacheName).mapPartitionToPrimaryAndBackups(split.index)
+            .map(_.asInstanceOf[TcpDiscoveryNode].socketAddresses()).flatten.map(_.getHostName).toList
     }
 
     def objectSql(typeName: String, sql: String, args: Any*): RDD[(K, V)] = {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f00a9e99/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ColocationTest.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ColocationTest.scala
b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ColocationTest.scala
new file mode 100644
index 0000000..a0814fa
--- /dev/null
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ColocationTest.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.examples
+
+import org.apache.ignite.configuration.IgniteConfiguration
+import org.apache.ignite.spark.IgniteContext
+import org.apache.spark.{SparkContext, SparkConf}
+
+object ColocationTest {
+    def main(args: Array[String]) {
+        val conf = new SparkConf().setAppName("Colocation test")
+        val sc = new SparkContext(conf)
+
+        val ignite = new IgniteContext[Int, Int](sc, () ⇒ new IgniteConfiguration())
+
+        // Search for lines containing "Ignite".
+        val cache = ignite.fromCache("partitioned")
+
+        cache.saveTuples(sc.parallelize((1 to 100000).toSeq, 48).map(i => (i, i)))
+
+        // Execute parallel sum.
+        println("Local sum: " + (1 to 100000).sum)
+        println("Distributed sum: " + cache.map(_._2).sum())
+    }
+}
\ No newline at end of file


Mime
View raw message