You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/10/21 21:45:36 UTC
[06/10] git commit: DocumentToMetadataProcessor utility processor
DocumentToMetadataProcessor utility processor
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/6cd97b8e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/6cd97b8e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/6cd97b8e
Branch: refs/heads/master
Commit: 6cd97b8e8a77c4a29c3f3079223a94a4ef41b877
Parents: 576b849
Author: Steve Blackmon <sb...@w2odigital.com>
Authored: Wed Oct 1 15:42:55 2014 -0500
Committer: Steve Blackmon <sb...@w2odigital.com>
Committed: Wed Oct 1 15:42:55 2014 -0500
----------------------------------------------------------------------
.../DatumFromMetadataAsDocumentProcessor.java | 17 +--
.../processor/DocumentToMetadataProcessor.java | 118 +++++++++++++++++++
2 files changed, 119 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/6cd97b8e/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
index 5e58bb0..9aea4c4 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
@@ -78,7 +78,7 @@ public class DatumFromMetadataAsDocumentProcessor implements StreamsProcessor, S
return result;
}
- Map<String, Object> metadata = asMap(metadataObjectNode);
+ Map<String, Object> metadata = DocumentToMetadataProcessor.asMap(metadataObjectNode);
if(entry == null || entry.getMetadata() == null)
return result;
@@ -128,19 +128,4 @@ public class DatumFromMetadataAsDocumentProcessor implements StreamsProcessor, S
this.elasticsearchClientManager.getClient().close();
}
- 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/6cd97b8e/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
new file mode 100644
index 0000000..804e1ac
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
@@ -0,0 +1,118 @@
+/*
+ * 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
+ *
+ * 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.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 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.apache.streams.jackson.StreamsJacksonMapper;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Uses index and type in metadata to populate current document into datums
+ */
+public class DocumentToMetadataProcessor implements StreamsProcessor, Serializable {
+
+ public final static String STREAMS_ID = "DatumFromMetadataProcessor";
+
+ private ElasticsearchClientManager elasticsearchClientManager;
+ private ElasticsearchReaderConfiguration config;
+
+ private ObjectMapper mapper;
+
+ public DocumentToMetadataProcessor() {
+ Config config = StreamsConfigurator.config.getConfig("elasticsearch");
+ this.config = ElasticsearchConfigurator.detectReaderConfiguration(config);
+ }
+
+ public DocumentToMetadataProcessor(Config config) {
+ this.config = ElasticsearchConfigurator.detectReaderConfiguration(config);
+ }
+
+ public DocumentToMetadataProcessor(ElasticsearchReaderConfiguration config) {
+ this.config = config;
+ }
+
+ @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);
+
+ if(entry == null || metadata == null)
+ return result;
+
+ entry.setMetadata(metadata);
+
+ result.add(entry);
+
+ return result;
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ this.elasticsearchClientManager = new ElasticsearchClientManager(config);
+ mapper = StreamsJacksonMapper.getInstance();
+ mapper.registerModule(new JsonOrgModule());
+
+ }
+
+ @Override
+ public void cleanUp() {
+ this.elasticsearchClientManager.getClient().close();
+ }
+
+ public static 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;
+ }
+}