ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [36/36] incubator-ignite git commit: IGNITE-389 - WIP
Date Wed, 20 May 2015 02:52:32 GMT
IGNITE-389 - WIP


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

Branch: refs/heads/ignite-389
Commit: 389ec79dff949fda2c7c16c84141b1b5f86793b5
Parents: edd1a95
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Tue May 19 19:51:43 2015 -0700
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Tue May 19 19:51:43 2015 -0700

----------------------------------------------------------------------
 modules/spark/pom.xml                           | 47 ++++++-----
 .../org/apache/ignite/spark/IgniteContext.scala | 85 +++++++++++++++-----
 .../org/apache/ignite/spark/IgniteRDD.scala     | 13 +--
 .../spark/examples/ExampleConfiguration.scala   | 41 ++++++++++
 .../spark/examples/IgniteProcessExample.scala   | 44 ++++++++++
 .../spark/examples/IgniteStoreExample.scala     | 40 +++++++++
 .../spark/util/SerializablePredicate2.scala     | 32 ++++++++
 .../org/apache/ignite/spark/util/using.scala    | 32 --------
 8 files changed, 248 insertions(+), 86 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/pom.xml
----------------------------------------------------------------------
diff --git a/modules/spark/pom.xml b/modules/spark/pom.xml
index 9d4ea86..aeae234 100644
--- a/modules/spark/pom.xml
+++ b/modules/spark/pom.xml
@@ -48,37 +48,40 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.ignite</groupId>
-            <artifactId>ignite-scalar</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
             <groupId>org.scala-lang</groupId>
             <artifactId>scala-library</artifactId>
-            <version>2.11.2</version>
+            <version>2.10.4</version>
         </dependency>
         <dependency>
-            <groupId>org.scalatest</groupId>
-            <artifactId>scalatest_2.11</artifactId>
-            <version>2.2.4</version>
-            <scope>test</scope>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.10</artifactId>
+            <version>1.3.1</version>
             <exclusions>
                 <exclusion>
-                    <groupId>org.scala-lang</groupId>
-                    <artifactId>scala-library</artifactId>
+                    <groupId>com.twitter</groupId>
+                    <artifactId>chill_2.11</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.twitter</groupId>
+                    <artifactId>chill-java</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.apache.spark</groupId>
-            <artifactId>spark-core_2.11</artifactId>
-            <version>1.3.1</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.spark</groupId>
-            <artifactId>spark-sql_2.11</artifactId>
-            <version>1.3.1</version>
-        </dependency>
+        <!--<dependency>-->
+            <!--<groupId>org.apache.spark</groupId>-->
+            <!--<artifactId>spark-sql_2.10</artifactId>-->
+            <!--<version>1.3.1</version>-->
+            <!--<exclusions>-->
+                <!--<exclusion>-->
+                    <!--<groupId>com.twitter</groupId>-->
+                    <!--<artifactId>chill_2.11</artifactId>-->
+                <!--</exclusion>-->
+                <!--<exclusion>-->
+                    <!--<groupId>com.twitter</groupId>-->
+                    <!--<artifactId>chill-java</artifactId>-->
+                <!--</exclusion>-->
+            <!--</exclusions>-->
+        <!--</dependency>-->
     </dependencies>
 
     <build>

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteContext.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteContext.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteContext.scala
index 0913605..56d2a05 100644
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteContext.scala
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/IgniteContext.scala
@@ -17,42 +17,51 @@
 
 package org.apache.ignite.spark
 
+import javax.cache.Cache
+
+import org.apache.ignite.cache.query.{Query, ScanQuery}
 import org.apache.ignite.cluster.ClusterNode
 import org.apache.ignite.internal.IgnitionEx
-import org.apache.ignite.spark.util.using
+import org.apache.ignite.lang.IgniteUuid
+import org.apache.ignite.spark.util.SerializablePredicate2
 import org.apache.ignite.{Ignition, IgniteCache, Ignite}
 import org.apache.ignite.configuration.{CacheConfiguration, IgniteConfiguration}
 import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 
+import scala.collection.JavaConversions._
+import scala.reflect.ClassTag
+
 class IgniteContext[K, V](
-    sc: SparkContext,
-    igniteCfg: IgniteConfiguration,
+    @scala.transient sc: SparkContext,
+    cfgF: () => IgniteConfiguration,
     val cacheName: String,
     cacheCfg: CacheConfiguration[K, V]
-) {
+) extends Serializable {
+    type ScanRDD[K1, V1] = IgniteRDD[Cache.Entry[K1, V1], K1, V1]
+
     def this(
         sc: SparkContext,
         springUrl: String,
         cacheName: String
     ) {
-        this(sc, IgnitionEx.loadConfiguration(springUrl).get1(), cacheName, null)
+        this(sc, () => IgnitionEx.loadConfiguration(springUrl).get1(), cacheName, null)
     }
 
     def this(
         sc: SparkContext,
-        igniteCfg: IgniteConfiguration,
+        cfgF: () => IgniteConfiguration,
         cacheName: String
     ) {
-        this(sc, igniteCfg, cacheName, null)
+        this(sc, cfgF, cacheName, null)
     }
 
     def this(
         sc: SparkContext,
-        igniteCfg: IgniteConfiguration,
+        cfgF: () => IgniteConfiguration,
         cacheCfg: CacheConfiguration[K, V]
     ) {
-        this(sc, igniteCfg, cacheCfg.getName, cacheCfg)
+        this(sc, cfgF, cacheCfg.getName, cacheCfg)
     }
 
     def this(
@@ -60,43 +69,75 @@ class IgniteContext[K, V](
         springUrl: String,
         cacheCfg: CacheConfiguration[K, V]
     ) {
-        this(sc, IgnitionEx.loadConfiguration(springUrl).get1(), cacheCfg.getName, cacheCfg)
+        this(sc, () => IgnitionEx.loadConfiguration(springUrl).get1(), cacheCfg.getName,
cacheCfg)
     }
 
     def sparkContext() = sc
 
-    def saveToIgnite(rdd: RDD[V], keyFunc: (IgniteContext[K, V], V, ClusterNode) => K
= affinityKeyFunc) = {
+    def scan(p: (K, V) => Boolean = (_, _) => true): ScanRDD[K, V] = {
+        new ScanRDD(this, new ScanQuery[K, V](new SerializablePredicate2[K, V](p)))
+    }
+
+    def scan[R:ClassTag](qry: Query[R]): IgniteRDD[R, K, V] = {
+        new IgniteRDD[R, K, V](this, qry)
+    }
+
+    def saveToIgnite[T](rdd: RDD[V], keyFunc: (IgniteContext[K, V], V, ClusterNode) =>
T = affinityKeyFunc(_: IgniteContext[K, V], _:V, _: ClusterNode)) = {
         rdd.foreachPartition(it => {
-            // TODO get affinity node
+            println("Using scala version: " + scala.util.Properties.versionString)
+            // Make sure to deploy the cache
+            igniteCache()
+
+            val ig = ignite()
 
-            using(ignite().dataStreamer[K, V](cacheName)) { streamer =>
+            val locNode = ig.cluster().localNode()
+
+            val node: Option[ClusterNode] = ig.cluster().forHost(locNode).nodes().find(!_.eq(locNode))
+
+            val streamer = ignite().dataStreamer[T, V](cacheName)
+
+            try {
                 it.foreach(value => {
-                    val key: K = keyFunc(this, value, null)
+                    val key: T = keyFunc(this, value, node.orNull)
+
+                    println("Saving: " + key + ", " + value)
+
                     streamer.addData(key, value)
                 })
             }
+            finally {
+                streamer.close()
+            }
         })
     }
 
     def ignite(): Ignite = {
+        val igniteCfg = cfgF()
+
         try {
             Ignition.ignite(igniteCfg.getGridName)
         }
         catch {
             case e: Exception =>
-                igniteCfg.setClientMode(true)
+                try {
+                    igniteCfg.setClientMode(true)
 
-                Ignition.start(igniteCfg)
+                    Ignition.start(igniteCfg)
+                }
+                catch {
+                    case e: Exception => Ignition.ignite(igniteCfg.getGridName)
+                }
         }
     }
 
-    def igniteCache(): IgniteCache[K, V] = {
-//        new IgniteRDD[Object, K, V](this, (k: K, v: V) => {true})
-
-        ignite().cache(cacheName)
+    private def igniteCache(): IgniteCache[K, V] = {
+        if (cacheCfg == null)
+            ignite().getOrCreateCache(cacheName)
+        else
+            ignite().getOrCreateCache(cacheCfg)
     }
 
-    private def affinityKeyFunc(ic: IgniteContext[K, V], key: K, node: ClusterNode) = {
-        null
+    private def affinityKeyFunc(ic: IgniteContext[K, V], value: V, node: ClusterNode): Object
= {
+        IgniteUuid.randomUuid()
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/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 186d1ae..4018c53 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,25 +17,18 @@
 
 package org.apache.ignite.spark
 
-import org.apache.ignite.cache.query.{ScanQuery, Query}
-import org.apache.ignite.scalar.lang.ScalarPredicate2
+import org.apache.ignite.cache.query.Query
 import org.apache.ignite.spark.impl.{IgniteQueryIterator, IgnitePartition}
 import org.apache.spark.{TaskContext, Partition}
 import org.apache.spark.rdd.RDD
 
 import scala.collection.JavaConversions._
+import scala.reflect.ClassTag
 
-class IgniteRDD[R, K, V](
+class IgniteRDD[R:ClassTag, K, V](
     ic: IgniteContext[K, V],
     qry: Query[R]
 ) extends RDD[R] (ic.sparkContext(), deps = Nil) {
-    def this(
-        ic: IgniteContext[K, V],
-        p: (K, V) => Boolean
-    ) = {
-        this(ic, new ScanQuery[K, V](new ScalarPredicate2[K, V](p)))
-    }
-
     override def compute(part: Partition, context: TaskContext): Iterator[R] = {
         new IgniteQueryIterator[R, K, V](ic, part, qry)
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ExampleConfiguration.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ExampleConfiguration.scala
b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ExampleConfiguration.scala
new file mode 100644
index 0000000..3b0dac7
--- /dev/null
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/ExampleConfiguration.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.internal.util.lang.{GridFunc => F}
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder
+
+object ExampleConfiguration {
+    def configuration(): IgniteConfiguration = {
+        val cfg = new IgniteConfiguration()
+
+        val discoSpi = new TcpDiscoverySpi()
+
+        val ipFinder = new TcpDiscoveryVmIpFinder()
+
+        ipFinder.setAddresses(F.asList("127.0.0.1:47500", "127.0.0.1:47501"))
+
+        discoSpi.setIpFinder(ipFinder)
+
+        cfg.setDiscoverySpi(discoSpi)
+
+        cfg
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteProcessExample.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteProcessExample.scala
b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteProcessExample.scala
new file mode 100644
index 0000000..4aeecb0
--- /dev/null
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteProcessExample.scala
@@ -0,0 +1,44 @@
+/*
+ * 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.spark.IgniteContext
+import org.apache.spark.{SparkContext, SparkConf}
+
+object IgniteProcessExample {
+    def main(args: Array[String]) {
+        val conf = new SparkConf().setAppName("Ignite processing example")
+        val sc = new SparkContext(conf)
+
+        val partitioned = new IgniteContext[Object, String](sc, ExampleConfiguration.configuration
_, "partitioned")
+
+        // Search for lines containing "Ignite".
+        val scan = partitioned.scan((k, v) => v.contains("Ignite"))
+
+        val processed = scan.filter(line => {
+            println("Analyzing line: " + line)
+
+            true
+        }).map(_.getValue)
+
+        // Create a new cache for results.
+        val results = new IgniteContext[Object, String](sc, ExampleConfiguration.configuration
_, "results")
+
+        results.saveToIgnite(processed)
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteStoreExample.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteStoreExample.scala
b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteStoreExample.scala
new file mode 100644
index 0000000..a7823f4
--- /dev/null
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/examples/IgniteStoreExample.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.spark.IgniteContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+import org.apache.spark.rdd.RDD
+
+object IgniteStoreExample {
+    def main(args: Array[String]) {
+        val conf = new SparkConf().setAppName("Ignite store example")
+        val sc = new SparkContext(conf)
+
+        val ignite = new IgniteContext[String, String](sc, ExampleConfiguration.configuration
_, "partitioned")
+
+        val lines: RDD[String] = sc.textFile(args(0)).filter(line => {
+            println("Read line: " + line)
+
+            true
+        })
+
+        ignite.saveToIgnite(lines)
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/util/SerializablePredicate2.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/util/SerializablePredicate2.scala
b/modules/spark/src/main/scala/org/apache/ignite/spark/util/SerializablePredicate2.scala
new file mode 100644
index 0000000..484d0df
--- /dev/null
+++ b/modules/spark/src/main/scala/org/apache/ignite/spark/util/SerializablePredicate2.scala
@@ -0,0 +1,32 @@
+/*
+ * 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.util
+
+import org.apache.ignite.lang.IgniteBiPredicate
+
+/**
+ * Peer deploy aware adapter for Java's `GridPredicate2`.
+ */
+class SerializablePredicate2[T1, T2](private val p: (T1, T2) => Boolean) extends IgniteBiPredicate[T1,
T2] {
+    assert(p != null)
+
+    /**
+     * Delegates to passed in function.
+     */
+    def apply(e1: T1, e2: T2) = p(e1, e2)
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/389ec79d/modules/spark/src/main/scala/org/apache/ignite/spark/util/using.scala
----------------------------------------------------------------------
diff --git a/modules/spark/src/main/scala/org/apache/ignite/spark/util/using.scala b/modules/spark/src/main/scala/org/apache/ignite/spark/util/using.scala
deleted file mode 100644
index 3b46d16..0000000
--- a/modules/spark/src/main/scala/org/apache/ignite/spark/util/using.scala
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.util
-
-import scala.util.Try
-
-object using {
-    type AutoClosable = { def close(): Unit }
-
-    def apply[A <: AutoClosable, B](resource: A)(code: A => B): B =
-        try {
-            code(resource)
-        }
-        finally {
-            Try(resource.close())
-        }
-}
\ No newline at end of file


Mime
View raw message