ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dsetrak...@apache.org
Subject incubator-ignite git commit: # ignite-45 - fixing streaming.
Date Thu, 19 Mar 2015 05:35:34 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-45 888f0a3d1 -> d49f9c885


# ignite-45 - fixing streaming.


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

Branch: refs/heads/ignite-45
Commit: d49f9c88506b39e384b2697a33f075ceaed66cef
Parents: 888f0a3
Author: Dmitiry Setrakyan <dsetrakyan@gridgain.com>
Authored: Thu Mar 19 01:35:27 2015 -0400
Committer: Dmitiry Setrakyan <dsetrakyan@gridgain.com>
Committed: Thu Mar 19 01:35:27 2015 -0400

----------------------------------------------------------------------
 .../ignite/examples/java8/ExamplesUtils.java    |  64 ++++++--
 .../StreamingPopularNumbersExample.java         | 163 -------------------
 .../java8/streaming/numbers/CacheConfig.java    |  33 ++++
 .../streaming/numbers/QueryPopularNumbers.java  |  67 ++++++++
 .../streaming/numbers/StreamRandomNumbers.java  |  79 +++++++++
 .../apache/ignite/stream/StreamTransformer.java |   9 +-
 6 files changed, 234 insertions(+), 181 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/examples/src/main/java8/org/apache/ignite/examples/java8/ExamplesUtils.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/ExamplesUtils.java b/examples/src/main/java8/org/apache/ignite/examples/java8/ExamplesUtils.java
index 5b62bd4..5854cdd 100644
--- a/examples/src/main/java8/org/apache/ignite/examples/java8/ExamplesUtils.java
+++ b/examples/src/main/java8/org/apache/ignite/examples/java8/ExamplesUtils.java
@@ -19,9 +19,9 @@ package org.apache.ignite.examples.java8;
 
 import org.apache.ignite.*;
 import org.apache.ignite.cluster.*;
-import org.apache.ignite.streamer.*;
 
 import java.net.*;
+import java.util.*;
 
 /**
  *
@@ -63,17 +63,15 @@ public class ExamplesUtils {
     /**
      * Checks minimum topology size for running a certain example.
      *
-     * @param prj Cluster to check size for.
+     * @param grp Cluster to check size for.
      * @param size Minimum number of nodes required to run a certain example.
      * @return {@code True} if check passed, {@code false} otherwise.
      */
-    public static boolean checkMinTopologySize(ClusterGroup prj, int size) {
-        int prjSize = prj.nodes().size();
+    public static boolean checkMinTopologySize(ClusterGroup grp, int size) {
+        int prjSize = grp.nodes().size();
 
         if (prjSize < size) {
-            System.out.println();
-            System.out.println(">>> Please start at least " + size + " cluster nodes
to run example.");
-            System.out.println();
+            System.err.println(">>> Please start at least " + size + " cluster nodes
to run example.");
 
             return false;
         }
@@ -82,18 +80,50 @@ public class ExamplesUtils {
     }
 
     /**
-     * @param ignite Ignite.
-     * @param name Streamer name.
-     * @return {@code True} if ignite has streamer with given name.
+     * Checks if cluster has server nodes.
+     *
+     * @param ignite Ignite instance.
+     * @return {@code True} if cluster has server nodes, {@code false} otherwise.
      */
-    public static boolean hasStreamer(Ignite ignite, String name) {
-        if (ignite.configuration().getStreamerConfiguration() != null) {
-            for (StreamerConfiguration cfg : ignite.configuration().getStreamerConfiguration())
{
-                if (name.equals(cfg.getName()))
-                    return true;
-            }
+    public static boolean hasServerNodes(Ignite ignite) {
+        if (ignite.cluster().forServers().nodes().isEmpty()) {
+            System.err.println("Server nodes not found (start data nodes with ExampleNodeStartup
class)");
+
+            return false;
         }
 
-        return false;
+        return true;
+    }
+
+    /**
+     * Convenience method for printing query results.
+     *
+     * @param res Query results.
+     */
+    public static void printQueryResults(List<?> res) {
+        if (res == null)
+            System.out.println("Query result set is empty.");
+        else {
+            System.out.println("Query results:");
+
+            for (Object row : res) {
+                if (row instanceof List) {
+                    System.out.print("  (");
+
+                    List<?> l = (List)row;
+
+                    for (int i = 0; i < l.size(); i++) {
+                        System.out.print(l.get(i));
+
+                        if (i + 1 != l.size())
+                            System.out.print(',');
+                    }
+
+                    System.out.println(')');
+                }
+                else
+                    System.out.println("  " + row);
+            }
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/StreamingPopularNumbersExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/StreamingPopularNumbersExample.java
b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/StreamingPopularNumbersExample.java
deleted file mode 100644
index 3b33402..0000000
--- a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/StreamingPopularNumbersExample.java
+++ /dev/null
@@ -1,163 +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.examples.java8.streaming;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.cache.query.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.examples.java8.*;
-import org.apache.ignite.stream.*;
-
-import javax.cache.*;
-import javax.cache.configuration.*;
-import javax.cache.expiry.*;
-import java.util.*;
-import java.util.concurrent.*;
-
-import static java.util.concurrent.TimeUnit.*;
-
-/**
- * Real time popular numbers counter.
- * <p>
- * Remote nodes should always be started with special configuration file which
- * enables P2P class loading: {@code 'ignite.{sh|bat} examples/config/example-compute.xml'}.
- * <p>
- * Alternatively you can run {@link ExampleNodeStartup} in another JVM which will
- * start node with {@code examples/config/example-compute.xml} configuration.
- */
-public class StreamingPopularNumbersExample {
-    /** Cache name. */
-    private static final String STREAM_NAME = StreamingPopularNumbersExample.class.getSimpleName();
-
-    /** Random number generator. */
-    private static final Random RAND = new Random();
-
-    /** Range within which to generate numbers. */
-    private static final int RANGE = 1000;
-
-    /** Test duration. */
-    private static final long DURATION = 2 * 60 * 1000;
-
-    /** Flag indicating that the test is finished. */
-    private static volatile boolean finished = false;
-
-    /**
-     * Executes example.
-     *
-     * @param args Command line arguments, none required.
-     * @throws IgniteException If example execution failed.
-     */
-    public static void main(String[] args) throws Exception {
-        // Mark this cluster member as client.
-        Ignition.setClientMode(true);
-
-        try (Ignite ignite = Ignition.start("examples/config/example-compute.xml")) {
-            System.out.println();
-            System.out.println(">>> Cache popular numbers example started.");
-
-            /*
-             * Configure streaming cache.
-             * =========================
-             */
-            CacheConfiguration<Integer, Long> cfg = new CacheConfiguration<>();
-
-            cfg.setCacheMode(CacheMode.PARTITIONED);
-            cfg.setName(STREAM_NAME);
-            cfg.setIndexedTypes(Integer.class, Long.class);
-
-            // Sliding window of 1 seconds.
-            cfg.setExpiryPolicyFactory(FactoryBuilder.factoryOf(new CreatedExpiryPolicy(new
Duration(SECONDS, 1))));
-
-            /**
-             * Start the streaming cache on all server nodes.
-             * ============================================
-             */
-            try (IgniteCache<Integer, Long> stmCache = ignite.createCache(cfg)) {
-                // Check that that server nodes have been started.
-                if (ignite.cluster().forDataNodes(STREAM_NAME).nodes().isEmpty()) {
-                    System.err.println("Data nodes not found (start data nodes with ExampleNodeStartup
class)");
-
-                    return;
-                }
-
-                ExecutorService exe = startStreaming(ignite);
-
-                long start = System.currentTimeMillis();
-
-                while (System.currentTimeMillis() - start < DURATION) {
-                    // Select top 10 words.
-                    SqlFieldsQuery top10 = new SqlFieldsQuery(
-                        "select _key, _val from Long order by _val desc limit 10");
-
-                    List<List<?>> results = stmCache.queryFields(top10).getAll();
-
-                    for (List<?> res : results)
-                        System.out.println(res.get(0) + "=" + res.get(1));
-
-                    System.out.println("----------------");
-
-                    Thread.sleep(5000);
-                }
-
-                finished = true;
-
-                exe.shutdown();
-            }
-            catch (CacheException e) {
-                e.printStackTrace();
-
-                System.out.println("Destroying cache for name '" + STREAM_NAME + "'. Please
try again.");
-
-                ignite.destroyCache(STREAM_NAME);
-            }
-        }
-    }
-
-    /**
-     * Populates the streaming cache in real time with numbers and keeps count for every
number.
-     *
-     * @param ignite Ignite.
-     */
-    private static ExecutorService startStreaming(final Ignite ignite) {
-        ExecutorService exe = Executors.newSingleThreadExecutor();
-
-        // Stream random numbers from another thread.
-        exe.submit(() -> {
-            try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(STREAM_NAME))
{
-                // Allow data updates.
-                stmr.allowOverwrite(true);
-
-                // Configure data transformation to count instances of the same word.
-                stmr.receiver(new StreamTransformer<>((e, args) -> {
-                    Long val = e.getValue();
-
-                    e.setValue(val == null ? 1L : val + 1);
-
-                    return null;
-                }));
-
-
-                while (!finished)
-                    stmr.addData(RAND.nextInt(RANGE), 1L);
-            }
-        });
-
-        return exe;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/CacheConfig.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/CacheConfig.java
b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/CacheConfig.java
new file mode 100644
index 0000000..76f50b1
--- /dev/null
+++ b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/CacheConfig.java
@@ -0,0 +1,33 @@
+package org.apache.ignite.examples.java8.streaming.numbers;
+
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+
+import javax.cache.configuration.*;
+import javax.cache.expiry.*;
+
+import static java.util.concurrent.TimeUnit.*;
+
+/**
+ * Created by Dmitriy on 3/18/15.
+ */
+public class CacheConfig {
+    /** Cache name. */
+    public static final String STREAM_NAME = "randomNumbers";
+
+    /**
+     * Configure streaming cache.
+     */
+    public static CacheConfiguration<Integer, Long> configure() {
+        CacheConfiguration<Integer, Long> cfg = new CacheConfiguration<>();
+
+        cfg.setCacheMode(CacheMode.PARTITIONED);
+        cfg.setName(STREAM_NAME);
+        cfg.setIndexedTypes(Integer.class, Long.class);
+
+        // Sliding window of 1 seconds.
+        cfg.setExpiryPolicyFactory(FactoryBuilder.factoryOf(new CreatedExpiryPolicy(new Duration(SECONDS,
1))));
+
+        return cfg;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/QueryPopularNumbers.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/QueryPopularNumbers.java
b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/QueryPopularNumbers.java
new file mode 100644
index 0000000..47be047
--- /dev/null
+++ b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/QueryPopularNumbers.java
@@ -0,0 +1,67 @@
+/*
+ * 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.examples.java8.streaming.numbers;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.query.*;
+import org.apache.ignite.examples.java8.*;
+
+import java.util.*;
+
+/**
+ * Real time popular numbers counter.
+ * <p>
+ * Remote nodes should always be started with special configuration file which
+ * enables P2P class loading: {@code 'ignite.{sh|bat} examples/config/example-compute.xml'}.
+ * <p>
+ * Alternatively you can run {@link ExampleNodeStartup} in another JVM which will
+ * start node with {@code examples/config/example-compute.xml} configuration.
+ */
+public class QueryPopularNumbers {
+    /**
+     * Executes example.
+     *
+     * @param args Command line arguments, none required.
+     * @throws IgniteException If example execution failed.
+     */
+    public static void main(String[] args) throws Exception {
+        // Mark this cluster member as client.
+        Ignition.setClientMode(true);
+
+        try (Ignite ignite = Ignition.start("examples/config/example-compute.xml")) {
+            // Start new cache or get existing one.
+            try (IgniteCache<Integer, Long> stmCache = ignite.createCache(CacheConfig.configure()))
{
+                if (!ExamplesUtils.hasServerNodes(ignite))
+                    return;
+
+                while (true) {
+                    // Select top 10 words.
+                    SqlFieldsQuery top10 = new SqlFieldsQuery(
+                        "select _key, _val from Long order by _val desc limit 10");
+
+                    // Execute query.
+                    List<List<?>> results = stmCache.queryFields(top10).getAll();
+
+                    ExamplesUtils.printQueryResults(results);
+
+                    Thread.sleep(5000);
+                }
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
new file mode 100644
index 0000000..96472a3
--- /dev/null
+++ b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
@@ -0,0 +1,79 @@
+/*
+ * 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.examples.java8.streaming.numbers;
+
+import org.apache.ignite.*;
+import org.apache.ignite.examples.java8.*;
+import org.apache.ignite.stream.*;
+
+import java.util.*;
+
+/**
+ * Real time popular numbers counter.
+ * <p>
+ * Remote nodes should always be started with special configuration file which
+ * enables P2P class loading: {@code 'ignite.{sh|bat} examples/config/example-compute.xml'}.
+ * <p>
+ * Alternatively you can run {@link ExampleNodeStartup} in another JVM which will
+ * start node with {@code examples/config/example-compute.xml} configuration.
+ */
+public class StreamRandomNumbers {
+    /** Random number generator. */
+    private static final Random RAND = new Random();
+
+    /** Range within which to generate numbers. */
+    private static final int RANGE = 1000;
+
+    /**
+     * Executes example.
+     *
+     * @param args Command line arguments, none required.
+     * @throws IgniteException If example execution failed.
+     */
+    public static void main(String[] args) throws Exception {
+        // Mark this cluster member as client.
+        Ignition.setClientMode(true);
+
+        try (Ignite ignite = Ignition.start("examples/config/example-compute.xml")) {
+            // Create new cache or get existing one.
+            try (IgniteCache<Integer, Long> stmCache = ignite.createCache(CacheConfig.configure()))
{
+                if (!ExamplesUtils.hasServerNodes(ignite))
+                    return;
+
+                try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(stmCache.getName()))
{
+                    // Allow data updates.
+                    stmr.allowOverwrite(true);
+
+                    // Configure data transformation to count instances of the same word.
+                    stmr.receiver(new StreamTransformer<>((e, arg) -> {
+                        Long val = e.getValue();
+
+                        e.setValue(val == null ? 1L : val + 1);
+
+                        return null;
+                    }));
+
+
+                    // Stream random numbers into the streamer cache.
+                    while (true)
+                        stmr.addData(RAND.nextInt(RANGE), 1L);
+                }
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d49f9c88/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
index 4ff9a59..ea95bf1 100644
--- a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
+++ b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
@@ -24,11 +24,18 @@ import javax.cache.processor.*;
 import java.util.*;
 
 /**
- * Created by Dmitriy on 3/18/15.
+ * Convenience adapter to transform update existing values in streaming cache
+ * based on the previously cached value.
  */
 public class StreamTransformer<K, V> implements StreamReceiver<K, V> {
+    /** Entry processor. */
     private EntryProcessor<K, V, Object> ep;
 
+    /**
+     * Entry processor to update cache values based on the previously cached value.
+     *
+     * @param ep Entry processor.
+     */
     public StreamTransformer(CacheEntryProcessor<K, V, Object> ep) {
         this.ep = ep;
     }


Mime
View raw message