streams-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sblack...@apache.org
Subject git commit: new examples, new ES processors to make them possible
Date Fri, 05 Sep 2014 19:56:15 GMT
Repository: incubator-streams
Updated Branches:
  refs/heads/pp ebfc4b4e7 -> d3ca0e7d0


new examples, new ES processors to make them possible


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

Branch: refs/heads/pp
Commit: d3ca0e7d0f6c0ea4bd09fbd8f62f45d45fa068e0
Parents: ebfc4b4
Author: sblackmon <sblackmon@apache.org>
Authored: Fri Sep 5 14:55:48 2014 -0500
Committer: sblackmon <sblackmon@apache.org>
Committed: Fri Sep 5 14:55:48 2014 -0500

----------------------------------------------------------------------
 .../ElasticsearchPersistDeleter.java            | 31 ++++---
 .../ElasticsearchPersistWriter.java             |  2 +-
 .../processor/DatumFromMetadataProcessor.java   | 92 ++++++++++++++++++++
 .../processor/MetadataAsDocumentProcessor.java  | 69 +++++++++++++++
 .../peoplepattern/AccountTypeProcessor.java     | 57 +++++++-----
 .../peoplepattern/DemographicsProcessor.java    | 56 +++++++-----
 .../streams/peoplepattern/AccountType.json      |  1 +
 7 files changed, 248 insertions(+), 60 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
index fece72e..7d74959 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
@@ -28,6 +28,8 @@ import org.elasticsearch.action.update.UpdateRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 public class ElasticsearchPersistDeleter extends ElasticsearchPersistWriter implements StreamsPersistWriter
{
 
     private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistDeleter.class);
@@ -44,21 +46,22 @@ public class ElasticsearchPersistDeleter extends ElasticsearchPersistWriter
impl
     public void write(StreamsDatum streamsDatum) {
 
         Preconditions.checkNotNull(streamsDatum);
-        Preconditions.checkNotNull(streamsDatum.getDocument());
         Preconditions.checkNotNull(streamsDatum.getMetadata());
-        Preconditions.checkNotNull(streamsDatum.getMetadata().get("id"));
-
-        String index;
-        String type;
-        String id;
-
-        index = Optional.fromNullable(
-                (String) streamsDatum.getMetadata().get("index"))
-                .or(config.getIndex());
-        type = Optional.fromNullable(
-                (String) streamsDatum.getMetadata().get("type"))
-                .or(config.getType());
-        id = (String) streamsDatum.getMetadata().get("id");
+
+        LOGGER.debug("Delete Metadata: {}", streamsDatum.getMetadata());
+
+        String index = (String) streamsDatum.getMetadata().get("index");
+        String type = (String) streamsDatum.getMetadata().get("type");
+        String id = setId(streamsDatum);
+
+        if(index == null || (config.getForceUseConfig() != null && config.getForceUseConfig()))
{
+            index = config.getIndex();
+        }
+        if(type == null || (config.getForceUseConfig() != null && config.getForceUseConfig()))
{
+            type = config.getType();
+        }
+
+        LOGGER.debug("Attempt Delete: ({},{},{})", index, type, id);
 
         delete(index, type, id);
 

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
index 664dd24..8300c62 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
@@ -164,7 +164,7 @@ public class ElasticsearchPersistWriter implements StreamsPersistWriter,
DatumSt
         }
     }
 
-    private String setId(StreamsDatum streamsDatum) {
+    protected String setId(StreamsDatum streamsDatum) {
         String id = Optional.fromNullable(
                 (String) streamsDatum.getMetadata().get("id"))
                 .orNull();

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
new file mode 100644
index 0000000..5e58190
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
@@ -0,0 +1,92 @@
+package org.apache.streams.elasticsearch.processor;
+
+import com.google.common.collect.Lists;
+import com.typesafe.config.Config;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.elasticsearch.ElasticsearchClientManager;
+import org.apache.streams.elasticsearch.ElasticsearchConfigurator;
+import org.apache.streams.elasticsearch.ElasticsearchReaderConfiguration;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetRequestBuilder;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.index.engine.Engine;
+import org.joda.time.DateTime;
+
+import java.util.List;
+
+/**
+ * Created by sblackmon on 9/4/14.
+ */
+public class DatumFromMetadataProcessor implements StreamsProcessor {
+
+    public final static String STREAMS_ID = "DatumFromMetadataProcessor";
+
+    private ElasticsearchClientManager elasticsearchClientManager;
+    private ElasticsearchReaderConfiguration config;
+
+    public DatumFromMetadataProcessor() {
+        Config config = StreamsConfigurator.config.getConfig("elasticsearch");
+        this.config = ElasticsearchConfigurator.detectReaderConfiguration(config);
+    }
+
+    public DatumFromMetadataProcessor(Config config) {
+        this.config = ElasticsearchConfigurator.detectReaderConfiguration(config);
+    }
+
+    public DatumFromMetadataProcessor(ElasticsearchReaderConfiguration config) {
+        this.config = config;
+    }
+
+    @Override
+    public List<StreamsDatum> process(StreamsDatum entry) {
+        List<StreamsDatum> result = Lists.newArrayList();
+
+        if(entry == null || entry.getMetadata() == null)
+            return result;
+
+        String index = (String) entry.getMetadata().get("index");
+        String type = (String) entry.getMetadata().get("type");
+        String id = (String) entry.getMetadata().get("id");
+
+        if( index == null ) {
+            index = this.config.getIndexes().get(0);
+        }
+        if( type == null ) {
+            type = this.config.getTypes().get(0);
+        }
+        if( id == null ) {
+            id = entry.getId();
+        }
+
+        GetRequestBuilder getRequestBuilder = elasticsearchClientManager.getClient().prepareGet(index,
type, id);
+        getRequestBuilder.setFields("*", "_timestamp");
+        getRequestBuilder.setFetchSource(true);
+        GetResponse getResponse = getRequestBuilder.get();
+
+        if( getResponse == null || getResponse.isExists() == false || getResponse.isSourceEmpty()
== true )
+            return result;
+
+        entry.setDocument(getResponse.getSource());
+        if( getResponse.getField("_timestamp") != null) {
+            DateTime timestamp = new DateTime(((Long) getResponse.getField("_timestamp").getValue()).longValue());
+            entry.setTimestamp(timestamp);
+        }
+
+        result.add(entry);
+
+        return result;
+    }
+
+    @Override
+    public void prepare(Object configurationObject) {
+        this.elasticsearchClientManager = new ElasticsearchClientManager(config);
+
+    }
+
+    @Override
+    public void cleanUp() {
+        this.elasticsearchClientManager.getClient().close();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataAsDocumentProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataAsDocumentProcessor.java
b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataAsDocumentProcessor.java
new file mode 100644
index 0000000..7a67760
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataAsDocumentProcessor.java
@@ -0,0 +1,69 @@
+package org.apache.streams.elasticsearch.processor;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by sblackmon on 9/5/14.
+ */
+public class MetadataAsDocumentProcessor implements StreamsProcessor {
+
+    public final static String STREAMS_ID = "DocumentToMetadataProcessor";
+
+    private ObjectMapper mapper;
+
+    @Override
+    public void prepare(Object configurationObject) {
+        mapper = StreamsJacksonMapper.getInstance();
+        mapper.registerModule(new JsonOrgModule());
+    }
+
+    @Override
+    public void cleanUp() {
+
+    }
+
+    @Override
+    public List<StreamsDatum> process(StreamsDatum entry) {
+        List<StreamsDatum> result = Lists.newArrayList();
+        ObjectNode metadataObjectNode;
+        try {
+            metadataObjectNode = mapper.readValue((String) entry.getDocument(), ObjectNode.class);
+        } catch (IOException e) {
+            return result;
+        }
+
+        Map<String, Object> metadata = asMap(metadataObjectNode);
+        entry.setMetadata(metadata);
+        result.add(entry);
+        return result;
+    }
+
+    public Map<String, Object> asMap(JsonNode node) {
+
+        Iterator<Map.Entry<String, JsonNode>> iterator = node.fields();
+        Map<String, Object> ret = Maps.newHashMap();
+
+        Map.Entry<String, JsonNode> entry;
+
+        while (iterator.hasNext()) {
+            entry = iterator.next();
+            if( entry.getValue().asText() != null )
+                ret.put(entry.getKey(), entry.getValue().asText());
+        }
+
+        return ret;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/AccountTypeProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/AccountTypeProcessor.java
b/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/AccountTypeProcessor.java
index e412639..cf5e853 100644
--- a/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/AccountTypeProcessor.java
+++ b/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/AccountTypeProcessor.java
@@ -57,7 +57,7 @@ public class AccountTypeProcessor implements StreamsProcessor {
 
     private final static Logger LOGGER = LoggerFactory.getLogger(AccountTypeProcessor.class);
 
-    private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+    private ObjectMapper mapper;
 
     private PeoplePatternConfiguration peoplePatternConfiguration = null;
 
@@ -115,34 +115,17 @@ public class AccountTypeProcessor implements StreamsProcessor {
         CloseableHttpClient httpclient = HttpClients.createDefault();
 
         CloseableHttpResponse response = null;
+
+        String entityString = null;
         try {
-            //response = httpclient.execute(httpget, context);
             response = httpclient.execute(httpget);
             HttpEntity entity = response.getEntity();
             // TODO: handle rate-limiting
             if (response.getStatusLine().getStatusCode() == 200 && entity != null)
{
-                String entityString = EntityUtils.toString(entity);
-
-                LOGGER.debug(entityString);
-
-                AccountType accountType = mapper.readValue(entityString, AccountType.class);
-
-                Map<String, Object> extensions = ActivityUtil.ensureExtensions(mapper.convertValue(actor,
ObjectNode.class));
-
-                extensions.put(EXTENSION, accountType);
-
-                actor.setAdditionalProperty(ActivityUtil.EXTENSION_PROPERTY, extensions);
-
-                LOGGER.debug("Actor: {}", actor);
-
-                activity.setActor(actor);
-
-                entry.setDocument(activity);
-
-                result.add(entry);
+                entityString = EntityUtils.toString(entity);
             }
         } catch (IOException e) {
-            LOGGER.error("IO error {} - {}", uri.toString(), response);
+            LOGGER.error("IO error:\n{}\n{}\n{}", uri.toString(), response, e.getMessage());
             return result;
         } finally {
             try {
@@ -153,12 +136,41 @@ public class AccountTypeProcessor implements StreamsProcessor {
             } catch (IOException e) {}
         }
 
+        if( entityString == null )
+            return result;
+
+        LOGGER.debug(entityString);
+
+        AccountType accountType;
+        try {
+            accountType = mapper.readValue(entityString, AccountType.class);
+        } catch (IOException e) {
+            LOGGER.error("IO error:\n{}\n{}\n{}", uri.toString(), response, e.getMessage());
+            return result;
+        }
+
+        Map<String, Object> extensions = ActivityUtil.ensureExtensions(mapper.convertValue(actor,
ObjectNode.class));
+
+        extensions.put(EXTENSION, accountType);
+
+        actor.setAdditionalProperty(ActivityUtil.EXTENSION_PROPERTY, extensions);
+
+        LOGGER.debug("Actor: {}", actor);
+
+        activity.setActor(actor);
+
+        entry.setDocument(activity);
+
+        result.add(entry);
+
+
         return result;
 
     }
 
     @Override
     public void prepare(Object configurationObject) {
+        mapper = StreamsJacksonMapper.getInstance();
         // TODO: one client object
         StringBuilder stringBuilder = new StringBuilder();
         stringBuilder.append(peoplePatternConfiguration.getUsername());
@@ -166,7 +178,6 @@ public class AccountTypeProcessor implements StreamsProcessor {
         stringBuilder.append(peoplePatternConfiguration.getPassword());
         String string = stringBuilder.toString();
         authHeader = Base64.encodeBase64String(string.getBytes());
-
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/DemographicsProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/DemographicsProcessor.java
b/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/DemographicsProcessor.java
index bd7a1b2..b87ad9a 100644
--- a/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/DemographicsProcessor.java
+++ b/streams-contrib/streams-processor-peoplepattern/src/main/java/org/apache/streams/peoplepattern/DemographicsProcessor.java
@@ -51,13 +51,13 @@ import java.util.Map;
  */
 public class DemographicsProcessor implements StreamsProcessor {
 
-    private final static String STREAMS_ID = "DemographicsProcessor";
+    public final static String STREAMS_ID = "DemographicsProcessor";
 
     private final static String EXTENSION = "demographics";
 
     private final static Logger LOGGER = LoggerFactory.getLogger(DemographicsProcessor.class);
 
-    private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+    private ObjectMapper mapper;
 
     private PeoplePatternConfiguration peoplePatternConfiguration = null;
 
@@ -110,33 +110,17 @@ public class DemographicsProcessor implements StreamsProcessor {
         CloseableHttpClient httpclient = HttpClients.createDefault();
 
         CloseableHttpResponse response = null;
+
+        String entityString = null;
         try {
             response = httpclient.execute(httpget);
             HttpEntity entity = response.getEntity();
             // TODO: handle rate-limiting
             if (response.getStatusLine().getStatusCode() == 200 && entity != null)
{
-                String entityString = EntityUtils.toString(entity);
-
-                LOGGER.debug(entityString);
-
-                Demographics demographics = mapper.readValue(entityString, Demographics.class);
-
-                Map<String, Object> extensions = ActivityUtil.ensureExtensions(mapper.convertValue(actor,
ObjectNode.class));
-
-                extensions.put(EXTENSION, demographics);
-
-                actor.setAdditionalProperty(ActivityUtil.EXTENSION_PROPERTY, extensions);
-
-                LOGGER.debug("Actor: {}", actor);
-
-                activity.setActor(actor);
-
-                entry.setDocument(activity);
-
-                result.add(entry);
+                entityString = EntityUtils.toString(entity);
             }
         } catch (IOException e) {
-            LOGGER.error("IO error {} - {}", uri.toString(), response);
+            LOGGER.error("IO error:\n{}\n{}\n{}", uri.toString(), response, e.getMessage());
             return result;
         } finally {
             try {
@@ -147,12 +131,40 @@ public class DemographicsProcessor implements StreamsProcessor {
             } catch (IOException e) {}
         }
 
+        if( entityString == null )
+            return result;
+
+        LOGGER.debug(entityString);
+
+        Demographics demographics = null;
+        try {
+            demographics = mapper.readValue(entityString, Demographics.class);
+        } catch (IOException e) {
+            LOGGER.error("IO error:\n{}\n{}\n{}", uri.toString(), response, e.getMessage());
+            return result;
+        }
+
+        Map<String, Object> extensions = ActivityUtil.ensureExtensions(mapper.convertValue(actor,
ObjectNode.class));
+
+        extensions.put(EXTENSION, demographics);
+
+        actor.setAdditionalProperty(ActivityUtil.EXTENSION_PROPERTY, extensions);
+
+        LOGGER.debug("Actor: {}", actor);
+
+        activity.setActor(actor);
+
+        entry.setDocument(activity);
+
+        result.add(entry);
+
         return result;
 
     }
 
     @Override
     public void prepare(Object configurationObject) {
+        mapper = StreamsJacksonMapper.getInstance();
         // TODO: one client object
         StringBuilder stringBuilder = new StringBuilder();
         stringBuilder.append(peoplePatternConfiguration.getUsername());

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d3ca0e7d/streams-contrib/streams-processor-peoplepattern/src/main/jsonschema/org/apache/streams/peoplepattern/AccountType.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-peoplepattern/src/main/jsonschema/org/apache/streams/peoplepattern/AccountType.json
b/streams-contrib/streams-processor-peoplepattern/src/main/jsonschema/org/apache/streams/peoplepattern/AccountType.json
index 61ff110..5656b44 100644
--- a/streams-contrib/streams-processor-peoplepattern/src/main/jsonschema/org/apache/streams/peoplepattern/AccountType.json
+++ b/streams-contrib/streams-processor-peoplepattern/src/main/jsonschema/org/apache/streams/peoplepattern/AccountType.json
@@ -12,6 +12,7 @@
                 "organization",
                 "entertainment",
                 "adult",
+                "spam",
                 "no-prediction"
             ],
             "default": "no-prediction"


Mime
View raw message