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 2015/03/20 01:54:13 UTC
[47/50] [abbrv] incubator-streams git commit: implements ordered
fields and metadata needs tests
implements ordered fields and metadata
needs tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/8d2b3e21
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/8d2b3e21
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/8d2b3e21
Branch: refs/heads/asf-master
Commit: 8d2b3e219fd1ba82078bb0cb09c37595cd0e0d00
Parents: aac4863
Author: sblackmon <sb...@apache.org>
Authored: Tue Mar 3 19:55:03 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Mar 3 19:55:03 2015 -0600
----------------------------------------------------------------------
streams-contrib/streams-persist-hdfs/pom.xml | 4 +-
.../org/apache/streams/hdfs/HdfsConstants.java | 13 ++++
.../streams/hdfs/WebHdfsPersistReader.java | 2 +-
.../streams/hdfs/WebHdfsPersistReaderTask.java | 78 ++++++++++++++------
.../streams/hdfs/WebHdfsPersistWriter.java | 45 +++++++----
.../apache/streams/hdfs/HdfsConfiguration.json | 20 +++++
6 files changed, 123 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/pom.xml b/streams-contrib/streams-persist-hdfs/pom.xml
index d924fb8..9bf86ed 100644
--- a/streams-contrib/streams-persist-hdfs/pom.xml
+++ b/streams-contrib/streams-persist-hdfs/pom.xml
@@ -115,9 +115,7 @@
<addCompileSourceRoot>true</addCompileSourceRoot>
<generateBuilders>true</generateBuilders>
<sourcePaths>
- <sourcePath>src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json</sourcePath>
- <sourcePath>src/main/jsonschema/org/apache/streams/hdfs/HdfsWriterConfiguration.json</sourcePath>
- <sourcePath>src/main/jsonschema/org/apache/streams/hdfs/HdfsReaderConfiguration.json</sourcePath>
+ <sourcePath>src/main/jsonschema</sourcePath>
</sourcePaths>
<outputDirectory>target/generated-sources/jsonschema2pojo</outputDirectory>
<targetPackage>org.apache.streams.hdfs.pojo</targetPackage>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/HdfsConstants.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/HdfsConstants.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/HdfsConstants.java
new file mode 100644
index 0000000..ab0ca1a
--- /dev/null
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/HdfsConstants.java
@@ -0,0 +1,13 @@
+package org.apache.streams.hdfs;
+
+/**
+ * Created by sblackmon on 3/3/15.
+ */
+public class HdfsConstants {
+
+ protected static final String ID = "ID";
+ protected static final String TS = "TS";
+ protected static final String META = "META";
+ protected static final String DOC = "DOC";
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
index b002d1a..02345a2 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
@@ -64,7 +64,7 @@ public class WebHdfsPersistReader implements StreamsPersistReader, DatumStatusCo
protected ObjectMapper mapper = StreamsJacksonMapper.getInstance();
- private HdfsReaderConfiguration hdfsConfiguration;
+ protected HdfsReaderConfiguration hdfsConfiguration;
private ExecutorService executor;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
index 7475d0d..3f9b906 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReaderTask.java
@@ -18,7 +18,9 @@
package org.apache.streams.hdfs;
+import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Strings;
+import com.google.common.collect.Maps;
import org.apache.hadoop.fs.FileStatus;
import org.apache.streams.core.DatumStatus;
import org.apache.streams.core.StreamsDatum;
@@ -28,7 +30,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
+import java.io.IOException;
import java.io.InputStreamReader;
+import java.util.Map;
public class WebHdfsPersistReaderTask implements Runnable {
@@ -102,34 +106,66 @@ public class WebHdfsPersistReaderTask implements Runnable {
private StreamsDatum processLine(String line) {
- String[] fields = line.split(Character.toString(reader.DELIMITER));
+ StreamsDatum datum;
+
+ String[] fields = line.split(reader.hdfsConfiguration.getFieldDelimiter());
if( fields.length == 0)
return null;
- if( fields.length == 1)
- return new StreamsDatum(fields[0]);
- if( fields.length == 2)
- return new StreamsDatum(fields[1], fields[0]);
- if( fields.length == 3) {
-
- DateTime timestamp = null;
- try {
- long longts = Long.parseLong(fields[1]);
- timestamp = new DateTime(longts);
- } catch ( Exception e ) {}
- try {
- timestamp = reader.mapper.readValue(fields[1], DateTime.class);
- } catch ( Exception e ) {}
-
- if( timestamp == null )
- return new StreamsDatum(fields[0], fields[2]);
- else
- return new StreamsDatum(fields[0], fields[2], timestamp);
+ String id;
+ DateTime ts;
+ Map<String, Object> metadata;
+ String json;
+
+ if( reader.hdfsConfiguration.getFields().contains( HdfsConstants.DOC )) {
+ json = fields[reader.hdfsConfiguration.getFields().indexOf(HdfsConstants.DOC)];
+ datum = new StreamsDatum(json);
+ } else {
+ return null;
+ }
+
+ if( reader.hdfsConfiguration.getFields().contains( HdfsConstants.ID ) ) {
+ id = fields[reader.hdfsConfiguration.getFields().indexOf(HdfsConstants.ID)];
+ datum.setId(id);
+ }
+ if( reader.hdfsConfiguration.getFields().contains( HdfsConstants.TS )) {
+ ts = parseTs(fields[reader.hdfsConfiguration.getFields().indexOf(HdfsConstants.TS)]);
+ datum.setTimestamp(ts);
+ }
+ if( reader.hdfsConfiguration.getFields().contains( HdfsConstants.META )) {
+ metadata = parseMap(fields[reader.hdfsConfiguration.getFields().indexOf(HdfsConstants.META)]);
+ datum.setMetadata(metadata);
}
- return new StreamsDatum(fields[3], fields[0]);
+ return datum;
}
+ private DateTime parseTs(String field) {
+
+ DateTime timestamp = null;
+ try {
+ long longts = Long.parseLong(field);
+ timestamp = new DateTime(longts);
+ } catch ( Exception e ) {}
+ try {
+ timestamp = reader.mapper.readValue(field, DateTime.class);
+ } catch ( Exception e ) {}
+
+ return timestamp;
+ }
+
+ private Map<String, Object> parseMap(String field) {
+
+ Map<String, Object> metadata = null;
+
+ try {
+ JsonNode jsonNode = reader.mapper.readValue(field, JsonNode.class);
+ metadata = reader.mapper.convertValue(jsonNode, Map.class);
+ } catch (IOException e) {
+ LOGGER.warn("failed in parseMap: " + e.getMessage());
+ }
+ return metadata;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
index 16b7f30..38ded28 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
@@ -20,7 +20,9 @@ package org.apache.streams.hdfs;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Joiner;
import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FileSystem;
@@ -40,6 +42,7 @@ import java.net.URISyntaxException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Date;
+import java.util.Iterator;
import java.util.List;
import java.util.Queue;
@@ -71,7 +74,7 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Flushable, Cl
private ObjectMapper mapper = new ObjectMapper();
- private HdfsWriterConfiguration hdfsConfiguration;
+ protected HdfsWriterConfiguration hdfsConfiguration;
public WebHdfsPersistWriter(HdfsWriterConfiguration hdfsConfiguration) {
this.hdfsConfiguration = hdfsConfiguration;
@@ -243,9 +246,9 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Flushable, Cl
}
private String convertResultToString(StreamsDatum entry) {
- String metadata = null;
+ String metadataJson = null;
try {
- metadata = mapper.writeValueAsString(entry.getMetadata());
+ metadataJson = mapper.writeValueAsString(entry.getMetadata());
} catch (JsonProcessingException e) {
LOGGER.warn("Error converting metadata to a string", e);
}
@@ -259,17 +262,31 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Flushable, Cl
if (Strings.isNullOrEmpty(documentJson))
return null;
- else
- return new StringBuilder()
- .append(entry.getId())
- .append(DELIMITER)
- .append(entry.getTimestamp())
- .append(DELIMITER)
- .append(metadata)
- .append(DELIMITER)
- .append(documentJson)
- .append("\n")
- .toString();
+ else {
+ StringBuilder stringBuilder = new StringBuilder();
+ Iterator<String> fields = hdfsConfiguration.getFields().iterator();
+ List<String> fielddata = Lists.newArrayList();
+ Joiner joiner = Joiner.on(hdfsConfiguration.getFieldDelimiter()).useForNull("");
+ while( fields.hasNext() ) {
+ String field = fields.next();
+ if( field.equals(HdfsConstants.DOC) )
+ fielddata.add(documentJson);
+ else if( field.equals(HdfsConstants.ID) )
+ fielddata.add(entry.getId());
+ else if( field.equals(HdfsConstants.TS) )
+ fielddata.add(entry.getTimestamp().toString());
+ else if( field.equals(HdfsConstants.META) )
+ fielddata.add(metadataJson);
+ else if( entry.getMetadata().containsKey(field)) {
+ fielddata.add(entry.getMetadata().get(field).toString());
+ } else {
+ fielddata.add(null);
+ }
+
+ }
+ joiner.appendTo(stringBuilder, fielddata);
+ return stringBuilder.append(hdfsConfiguration.getLineDelimiter()).toString();
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/8d2b3e21/streams-contrib/streams-persist-hdfs/src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json b/streams-contrib/streams-persist-hdfs/src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json
index 8d0a397..e6e1e4c 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json
+++ b/streams-contrib/streams-persist-hdfs/src/main/jsonschema/org/apache/streams/hdfs/HdfsConfiguration.json
@@ -33,6 +33,26 @@
"password": {
"type": "string",
"description": "Password"
+ },
+ "fields": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ },
+ "default": [
+ "ID",
+ "TS",
+ "META",
+ "DOC"
+ ]
+ },
+ "field_delimiter": {
+ "type": "string",
+ "default": "\t"
+ },
+ "line_delimiter": {
+ "type": "string",
+ "default": "\n"
}
}
}
\ No newline at end of file