You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by sa...@apache.org on 2016/05/10 09:05:55 UTC
[1/4] incubator-apex-malhar git commit: APEXMALHAR-2023 Adding
Enrichment Operator to Malhar
Repository: incubator-apex-malhar
Updated Branches:
refs/heads/master 3e09d7b25 -> 79c643b95
APEXMALHAR-2023 Adding Enrichment Operator to Malhar
Developed Map and Bean Enrichment Operators
Developed JDBC Loader
Developed HBase Loader
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/528e7aca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/528e7aca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/528e7aca
Branch: refs/heads/master
Commit: 528e7aca93ef32ce60be630d9d0f74ce3611059f
Parents: 08cb9d3
Author: Chaitanya <ch...@datatorrent.com>
Authored: Tue Mar 29 14:46:17 2016 +0530
Committer: chinmaykolhatkar <ch...@datatorrent.com>
Committed: Fri Apr 29 22:12:47 2016 +0530
----------------------------------------------------------------------
.../enrichment/AbstractEnrichmentOperator.java | 170 +++++++++++++++++
.../contrib/enrichment/EnrichmentBackup.java | 18 ++
.../contrib/enrichment/FSLoader.java | 146 +++++++++++++++
.../contrib/enrichment/HBaseLoader.java | 128 +++++++++++++
.../contrib/enrichment/JDBCLoader.java | 158 ++++++++++++++++
.../enrichment/MapEnrichmentOperator.java | 57 ++++++
.../enrichment/NullValuesCacheManager.java | 43 +++++
.../enrichment/POJOEnrichmentOperator.java | 185 +++++++++++++++++++
.../contrib/enrichment/ReadOnlyBackup.java | 38 ++++
.../contrib/enrichment/package-info.java | 1 +
.../enrichment/BeanEnrichmentOperatorTest.java | 95 ++++++++++
.../contrib/enrichment/EmployeeOrder.java | 95 ++++++++++
.../contrib/enrichment/FileEnrichmentTest.java | 75 ++++++++
.../contrib/enrichment/HBaseLoaderTest.java | 162 ++++++++++++++++
.../contrib/enrichment/JDBCLoaderTest.java | 179 ++++++++++++++++++
.../enrichment/MapEnrichmentOperatorTest.java | 152 +++++++++++++++
16 files changed, 1702 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
new file mode 100644
index 0000000..cbd4d5e
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
@@ -0,0 +1,170 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.lib.db.cache.CacheManager;
+import com.datatorrent.lib.db.cache.CacheStore;
+import com.datatorrent.lib.db.cache.CacheStore.ExpiryType;
+import com.esotericsoftware.kryo.NotNull;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Base class for Enrichment Operator. Subclasses should provide implementation to getKey and convert.
+ * The operator receives a tuple and emits enriched tuple based on includeFields and lookupFields. <br/>
+ *
+ * Properties:<br>
+ * <b>lookupFieldsStr</b>: List of comma separated keys for quick searching. Ex: Field1,Field2,Field3<br>
+ * <b>includeFieldsStr</b>: List of comma separated fields to be replaced/added to the input tuple. Ex: Field1,Field2,Field3<br>
+ * <b>store</b>: Specify the type of loader for looking data<br>
+ * <br>
+ *
+ *
+ * @displayName Abstract Enrichment Operator
+ * @tags Enrichment
+ * @param <INPUT> Type of tuples which are received by this operator</T>
+ * @param <OUTPUT> Type of tuples which are emitted by this operator</T>
+ * @since 2.1.0
+ */
+public abstract class AbstractEnrichmentOperator<INPUT, OUTPUT> extends BaseOperator
+{
+ /**
+ * Keep lookup data cache for fast access.
+ */
+ private transient CacheManager cacheManager;
+
+ private transient CacheStore primaryCache = new CacheStore();
+
+ private int entryExpiryDurationInMillis = 24 * 60 * 60 * 1000;
+ private int cacheCleanupInMillis = 24 * 60 * 60 * 1000;
+ private int cacheSize = 1024;
+
+ public transient DefaultOutputPort<OUTPUT> output = new DefaultOutputPort<OUTPUT>();
+
+ @InputPortFieldAnnotation(optional = true)
+ public transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
+ {
+ @Override public void process(INPUT tuple)
+ {
+ processTuple(tuple);
+ }
+ };
+
+ private EnrichmentBackup store;
+
+ @NotNull
+ protected String lookupFieldsStr;
+
+ protected String includeFieldsStr;
+
+ protected transient List<String> lookupFields = new ArrayList<String>();
+ protected transient List<String> includeFields = new ArrayList<String>();
+
+ protected void processTuple(INPUT tuple) {
+ Object key = getKey(tuple);
+ if(key != null) {
+ Object result = cacheManager.get(key);
+ OUTPUT out = convert(tuple, result);
+ emitTuple(out);
+ }
+ }
+
+ protected abstract Object getKey(INPUT tuple);
+
+ protected void emitTuple(OUTPUT tuple) {
+ output.emit(tuple);
+ }
+
+ /* Add data from cached value to input field */
+ protected abstract OUTPUT convert(INPUT in, Object cached);
+
+ @Override public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+
+ cacheManager = new NullValuesCacheManager();
+
+ // set expiration to one day.
+ primaryCache.setEntryExpiryDurationInMillis(entryExpiryDurationInMillis);
+ primaryCache.setCacheCleanupInMillis(cacheCleanupInMillis);
+ primaryCache.setEntryExpiryStrategy(ExpiryType.EXPIRE_AFTER_WRITE);
+ primaryCache.setMaxCacheSize(cacheSize);
+
+ lookupFields = Arrays.asList(lookupFieldsStr.split(","));
+ if (includeFieldsStr != null) {
+ includeFields = Arrays.asList(includeFieldsStr.split(","));
+ }
+
+ try {
+ store.setFields(lookupFields, includeFields);
+
+ cacheManager.setPrimary(primaryCache);
+ cacheManager.setBackup(store);
+ cacheManager.initialize();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to initialize primary cache", e);
+ }
+ }
+
+ /**
+ * Set the type of backup store for storing and searching data.
+ */
+ public void setStore(EnrichmentBackup store) {
+ this.store = store;
+ }
+
+ public EnrichmentBackup getStore() {
+ return store;
+ }
+
+ public CacheStore getPrimaryCache()
+ {
+ return primaryCache;
+ }
+
+ public String getLookupFieldsStr()
+ {
+ return lookupFieldsStr;
+ }
+
+ /**
+ * Set the lookup fields for quick searching. It would be in comma separated list
+ */
+ public void setLookupFieldsStr(String lookupFieldsStr)
+ {
+ this.lookupFieldsStr = lookupFieldsStr;
+ }
+
+ public String getIncludeFieldsStr()
+ {
+ return includeFieldsStr;
+ }
+
+ /**
+ * Set the list of comma separated fields to be added/replaced to the incoming tuple.
+ */
+ public void setIncludeFieldsStr(String includeFieldsStr)
+ {
+ this.includeFieldsStr = includeFieldsStr;
+ }
+
+ public void setEntryExpiryDurationInMillis(int entryExpiryDurationInMillis)
+ {
+ this.entryExpiryDurationInMillis = entryExpiryDurationInMillis;
+ }
+
+ public void setCacheCleanupInMillis(int cacheCleanupInMillis)
+ {
+ this.cacheCleanupInMillis = cacheCleanupInMillis;
+ }
+
+ public void setCacheSize(int cacheSize)
+ {
+ this.cacheSize = cacheSize;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
new file mode 100644
index 0000000..9155b13
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
@@ -0,0 +1,18 @@
+/**
+ * Copyright (c) 2015 DataTorrent, Inc.
+ * All rights reserved.
+ */
+
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.db.cache.CacheManager;
+import java.util.List;
+/**
+ * @since 3.1.0
+ */
+
+public interface EnrichmentBackup extends CacheManager.Backup
+{
+ public void setFields(List<String> lookupFields,List<String> includeFields);
+ public boolean needRefresh();
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
new file mode 100644
index 0000000..2effed0
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
@@ -0,0 +1,146 @@
+/**
+ * Copyright (c) 2015 DataTorrent, Inc.
+ * All rights reserved.
+ */
+package com.datatorrent.contrib.enrichment;
+
+import com.esotericsoftware.kryo.NotNull;
+import com.google.common.collect.Maps;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.codehaus.jackson.JsonProcessingException;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.ObjectReader;
+import org.codehaus.jackson.type.TypeReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * @since 3.1.0
+ */
+
+public class FSLoader extends ReadOnlyBackup
+{
+ @NotNull
+ private String fileName;
+
+ private transient Path filePath;
+ private transient FileSystem fs;
+ private transient boolean connected;
+
+ private transient static final ObjectMapper mapper = new ObjectMapper();
+ private transient static final ObjectReader reader = mapper.reader(new TypeReference<Map<String, Object>>()
+ {
+ });
+ private transient static final Logger logger = LoggerFactory.getLogger(FSLoader.class);
+
+ public String getFileName()
+ {
+ return fileName;
+ }
+
+ public void setFileName(String fileName)
+ {
+ this.fileName = fileName;
+ }
+
+ @Override public Map<Object, Object> loadInitialData()
+ {
+ Map<Object, Object> result = null;
+ FSDataInputStream in = null;
+ BufferedReader bin = null;
+ try {
+ result = Maps.newHashMap();
+ in = fs.open(filePath);
+ bin = new BufferedReader(new InputStreamReader(in));
+ String line;
+ while ((line = bin.readLine()) != null) {
+ try {
+ Map<String, Object> tuple = reader.readValue(line);
+ if(CollectionUtils.isEmpty(includeFields)) {
+ if(includeFields == null)
+ includeFields = new ArrayList<String>();
+ for (Map.Entry<String, Object> e : tuple.entrySet()) {
+ includeFields.add(e.getKey());
+ }
+ }
+ ArrayList<Object> includeTuple = new ArrayList<Object>();
+ for(String s: includeFields) {
+ includeTuple.add(tuple.get(s));
+ }
+ result.put(getKey(tuple), includeTuple);
+ } catch (JsonProcessingException parseExp) {
+ logger.info("Unable to parse line {}", line);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ if(bin != null)
+ IOUtils.closeQuietly(bin);
+ if(in != null)
+ IOUtils.closeQuietly(in);
+ try {
+ fs.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ logger.debug("loading initial data {}", result.size());
+ return result;
+ }
+
+ private Object getKey(Map<String, Object> tuple)
+ {
+ ArrayList<Object> lst = new ArrayList<Object>();
+ for(String key : lookupFields) {
+ lst.add(tuple.get(key));
+ }
+ return lst;
+ }
+
+ @Override public Object get(Object key)
+ {
+ return null;
+ }
+
+ @Override public List<Object> getAll(List<Object> keys)
+ {
+ return null;
+ }
+
+ @Override public void connect() throws IOException
+ {
+ Configuration conf = new Configuration();
+ this.filePath = new Path(fileName);
+ this.fs = FileSystem.newInstance(filePath.toUri(), conf);
+ if (!fs.isFile(filePath))
+ throw new IOException("Provided path " + fileName + " is not a file");
+ connected = true;
+ }
+
+ @Override public void disconnect() throws IOException
+ {
+ if (fs != null)
+ fs.close();
+ }
+
+ @Override public boolean isConnected()
+ {
+ return connected;
+ }
+
+ @Override
+ public boolean needRefresh() {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
new file mode 100755
index 0000000..7040a7a
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
@@ -0,0 +1,128 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.contrib.hbase.HBaseStore;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * <p>HBaseLoader extends from {@link HBaseStore} uses HBase to connect and implements EnrichmentBackup interface.</p> <br/>
+ *
+ * Properties:<br>
+ * <b>includeFamilys</b>: List of comma separated families and each family corresponds to the group name of column fields in includeFieldsStr. Ex: Family1,Family2<br>
+ * <br>
+ *
+ * @displayName HBaseLoader
+ * @tags Loader
+ * @since 2.1.0
+ */
+public class HBaseLoader extends HBaseStore implements EnrichmentBackup
+{
+ protected transient List<String> includeFields;
+ protected transient List<String> lookupFields;
+ protected transient List<String> includeFamilys;
+
+ protected Object getQueryResult(Object key)
+ {
+ try {
+ Get get = new Get(getRowBytes(((ArrayList)key).get(0)));
+ int idx = 0;
+ for(String f : includeFields) {
+ get.addColumn(Bytes.toBytes(includeFamilys.get(idx++)), Bytes.toBytes(f));
+ }
+ return getTable().get(get);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected ArrayList<Object> getDataFrmResult(Object result)
+ {
+ Result res = (Result)result;
+ if (res == null || res.isEmpty())
+ return null;
+ ArrayList<Object> columnInfo = new ArrayList<Object>();
+
+ if(CollectionUtils.isEmpty(includeFields)) {
+ if(includeFields == null) {
+ includeFields = new ArrayList<String>();
+ includeFamilys.clear();
+ includeFamilys = new ArrayList<String>();
+ }
+ for (KeyValue kv: res.raw()) {
+ includeFields.add(new String(kv.getQualifier()));
+ includeFamilys.add(new String(kv.getFamily()));
+ }
+ }
+ for(KeyValue kv : res.raw()) {
+ columnInfo.add(kv.getValue());
+ }
+ return columnInfo;
+ }
+
+ private byte[] getRowBytes(Object key)
+ {
+ return ((String)key).getBytes();
+ }
+
+ @Override public void setFields(List<String> lookupFields,List<String> includeFields)
+ {
+ this.includeFields = includeFields;
+ this.lookupFields = lookupFields;
+ }
+
+ /**
+ * Set the familyStr and would be in the form of comma separated list.
+ */
+ public void setIncludeFamilyStr(String familyStr)
+ {
+ this.includeFamilys = Arrays.asList(familyStr.split(","));
+ }
+
+ @Override public boolean needRefresh()
+ {
+ return false;
+ }
+
+ @Override public Map<Object, Object> loadInitialData()
+ {
+ return null;
+ }
+
+ @Override public Object get(Object key)
+ {
+ return getDataFrmResult(getQueryResult(key));
+ }
+
+ @Override public List<Object> getAll(List<Object> keys)
+ {
+ List<Object> values = Lists.newArrayList();
+ for (Object key : keys) {
+ values.add(get(key));
+ }
+ return values;
+ }
+
+ @Override public void put(Object key, Object value)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void putAll(Map<Object, Object> m)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void remove(Object key)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
new file mode 100644
index 0000000..3b1a8cf
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
@@ -0,0 +1,158 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.db.jdbc.JdbcStore;
+import com.google.common.collect.Lists;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.collections.CollectionUtils;
+
+/**
+ * <p>HBaseLoader extends from {@link JdbcStore} uses JDBC to connect and implements EnrichmentBackup interface.</p> <br/>
+ *
+ * Properties:<br>
+ * <b>queryStmt</b>: Sql Prepared Statement which needs to be executed<br>
+ * <b>tableName</b>: JDBC table name<br>
+ * <br>
+ *
+ * @displayName JDBCLoader
+ * @tags Loader
+ * @since 2.1.0
+ */
+public class JDBCLoader extends JdbcStore implements EnrichmentBackup
+{
+ protected String queryStmt;
+
+ protected String tableName;
+
+ protected transient List<String> includeFields;
+ protected transient List<String> lookupFields;
+
+ protected Object getQueryResult(Object key)
+ {
+ try {
+ PreparedStatement getStatement = getConnection().prepareStatement(queryStmt);
+ ArrayList<Object> keys = (ArrayList<Object>) key;
+ for (int i = 0; i < keys.size(); i++) {
+ getStatement.setObject(i+1, keys.get(i));
+ }
+ return getStatement.executeQuery();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected ArrayList<Object> getDataFrmResult(Object result) throws RuntimeException
+ {
+ try {
+ ResultSet resultSet = (ResultSet) result;
+ if (resultSet.next()) {
+ ResultSetMetaData rsdata = resultSet.getMetaData();
+ // If the includefields is empty, populate it from ResultSetMetaData
+ if(CollectionUtils.isEmpty(includeFields)) {
+ if(includeFields == null)
+ includeFields = new ArrayList<String>();
+ for (int i = 1; i <= rsdata.getColumnCount(); i++) {
+ includeFields.add(rsdata.getColumnName(i));
+ }
+ }
+ ArrayList<Object> res = new ArrayList<Object>();
+ for(String f : includeFields) {
+ res.add(resultSet.getObject(f));
+ }
+ return res;
+ } else
+ return null;
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private String generateQueryStmt()
+ {
+ String stmt = "select * from " + tableName + " where ";
+ for (int i = 0; i < lookupFields.size(); i++) {
+ stmt = stmt + lookupFields.get(i) + " = ? ";
+ if(i != lookupFields.size() - 1) {
+ stmt = stmt + " and ";
+ }
+ }
+ logger.info("generateQueryStmt: {}", stmt);
+ return stmt;
+ }
+
+ public String getQueryStmt()
+ {
+ return queryStmt;
+ }
+
+ @Override
+ public boolean needRefresh() {
+ return false;
+ }
+
+ /**
+ * Set the sql Prepared Statement if the enrichment mechanism is query based.
+ */
+ public void setQueryStmt(String queryStmt)
+ {
+ this.queryStmt = queryStmt;
+ }
+
+ public String getTableName()
+ {
+ return tableName;
+ }
+ /**
+ * Set the table name.
+ */
+ public void setTableName(String tableName)
+ {
+ this.tableName = tableName;
+ }
+
+ @Override public void setFields(List<String> lookupFields,List<String> includeFields)
+ {
+ this.includeFields = includeFields;
+ this.lookupFields = lookupFields;
+ if(queryStmt == null)
+ queryStmt = generateQueryStmt();
+ }
+ @Override public Map<Object, Object> loadInitialData()
+ {
+ return null;
+ }
+
+ @Override public Object get(Object key)
+ {
+ return getDataFrmResult(getQueryResult(key));
+ }
+
+ @Override public List<Object> getAll(List<Object> keys)
+ {
+ List<Object> values = Lists.newArrayList();
+ for (Object key : keys) {
+ values.add(get(key));
+ }
+ return values;
+ }
+
+ @Override public void put(Object key, Object value)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void putAll(Map<Object, Object> m)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void remove(Object key)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
new file mode 100644
index 0000000..040b5ae
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
@@ -0,0 +1,57 @@
+package com.datatorrent.contrib.enrichment;
+
+import java.util.ArrayList;
+import java.util.Map;
+
+/**
+ *
+ * This class takes a HashMap tuple as input and extract the value of the lookupKey configured
+ * for this operator. It then does a lookup in file/DB to find matching entry and all key-value pairs
+ * specified in the file/DB or based on include fields are added to original tuple.
+ *
+ * Example
+ * The file contains data in json format, one entry per line. during setup entire file is read and
+ * kept in memory for quick lookup.
+ * If file contains following lines, and operator is configured with lookup key "productId"
+ * { "productId": 1, "productCategory": 3 }
+ * { "productId": 4, "productCategory": 10 }
+ * { "productId": 3, "productCategory": 1 }
+ *
+ * And input tuple is
+ * { amount=10.0, channelId=4, productId=3 }
+ *
+ * The tuple is modified as below before operator emits it on output port.
+ * { amount=10.0, channelId=4, productId=3, productCategory=1 }
+ *
+ *
+ * @displayName MapEnrichment
+ * @category Database
+ * @tags enrichment, lookup
+ *
+ * @since 2.1.0
+ */
+public class MapEnrichmentOperator extends AbstractEnrichmentOperator<Map<String, Object>, Map<String, Object>>
+{
+ @Override protected Object getKey(Map<String, Object> tuple)
+ {
+ ArrayList<Object> keyList = new ArrayList<Object>();
+ for(String key : lookupFields) {
+ keyList.add(tuple.get(key));
+ }
+ return keyList;
+ }
+
+ @Override protected Map<String, Object> convert(Map<String, Object> in, Object cached)
+ {
+ if (cached == null)
+ return in;
+
+ ArrayList<Object> newAttributes = (ArrayList<Object>)cached;
+ if(newAttributes != null) {
+ for (int i = 0; i < includeFields.size(); i++) {
+ in.put(includeFields.get(i), newAttributes.get(i));
+ }
+ }
+ return in;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
new file mode 100644
index 0000000..f668683
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
@@ -0,0 +1,43 @@
+/**
+ * Copyright (c) 2015 DataTorrent, Inc.
+ * All rights reserved.
+ */
+
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.db.cache.CacheManager;
+
+/**
+ * @since 3.1.0
+ */
+public class NullValuesCacheManager extends CacheManager
+{
+
+ private static final NullObject NULL = new NullObject();
+ @Override
+ public Object get(Object key)
+ {
+ Object primaryVal = primary.get(key);
+ if (primaryVal != null) {
+ if (primaryVal == NULL) {
+ return null;
+ }
+
+ return primaryVal;
+ }
+
+ Object backupVal = backup.get(key);
+ if (backupVal != null) {
+ primary.put(key, backupVal);
+ } else {
+ primary.put(key, NULL);
+ }
+ return backupVal;
+
+ }
+
+ private static class NullObject
+ {
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
new file mode 100644
index 0000000..e707198
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
@@ -0,0 +1,185 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Getter;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.NotNull;
+import org.apache.commons.lang3.ClassUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * This class takes a POJO as input and extract the value of the lookupKey configured
+ * for this operator. It then does a lookup in file/DB to find matching entry and all key-value pairs
+ * specified in the file/DB or based on include fieldMap are added to original tuple.
+ *
+ * Properties:<br>
+ * <b>inputClass</b>: Class to be loaded for the incoming data type<br>
+ * <b>outputClass</b>: Class to be loaded for the emitted data type<br>
+ * <br>
+ *
+ * Example
+ * The file contains data in json format, one entry per line. during setup entire file is read and
+ * kept in memory for quick lookup.
+ * If file contains following lines, and operator is configured with lookup key "productId"
+ * { "productId": 1, "productCategory": 3 }
+ * { "productId": 4, "productCategory": 10 }
+ * { "productId": 3, "productCategory": 1 }
+ *
+ * And input tuple is
+ * { amount=10.0, channelId=4, productId=3 }
+ *
+ * The tuple is modified as below before operator emits it on output port.
+ * { amount=10.0, channelId=4, productId=3, productCategory=1 }
+ *
+ * @displayName BeanEnrichment
+ * @category Database
+ * @tags enrichment, lookup
+ *
+ * @since 2.1.0
+ */
+public class POJOEnrichmentOperator extends AbstractEnrichmentOperator<Object, Object> {
+
+ private transient static final Logger logger = LoggerFactory.getLogger(POJOEnrichmentOperator.class);
+ protected Class inputClass;
+ protected Class outputClass;
+ private transient List<Getter> getters = new LinkedList<Getter>();
+ private transient List<FieldObjectMap> fieldMap = new LinkedList<FieldObjectMap>();
+ private transient List<Setter> updateSetter = new LinkedList<Setter>();
+
+ @NotNull
+ protected String outputClassStr;
+
+
+ @Override
+ protected Object getKey(Object tuple) {
+ ArrayList<Object> keyList = new ArrayList<Object>();
+ for(Getter g : getters) {
+ keyList.add(g.get(tuple));
+ }
+ return keyList;
+ }
+
+ @Override
+ protected Object convert(Object in, Object cached) {
+ try {
+ Object o = outputClass.newInstance();
+
+ // Copy the fields from input to output
+ for (FieldObjectMap map : fieldMap) {
+ map.set.set(o, map.get.get(in));
+ }
+
+ if (cached == null)
+ return o;
+
+ if(updateSetter.size() == 0 && includeFields.size() != 0) {
+ populateUpdatesFrmIncludeFields();
+ }
+ ArrayList<Object> newAttributes = (ArrayList<Object>)cached;
+ int idx = 0;
+ for(Setter s: updateSetter) {
+ s.set(o, newAttributes.get(idx));
+ idx++;
+ }
+ return o;
+ } catch (InstantiationException e) {
+ throw new RuntimeException(e);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context) {
+ super.setup(context);
+ populateUpdatesFrmIncludeFields();
+ }
+
+ private void populateGettersFrmLookup()
+ {
+ for (String fName : lookupFields) {
+ Getter f = PojoUtils.createGetter(inputClass, fName, Object.class);
+ getters.add(f);
+ }
+ }
+
+ private void populateGettersFrmInput()
+ {
+ Field[] fields = inputClass.getFields();
+ for (Field f : fields) {
+ Class c = ClassUtils.primitiveToWrapper(f.getType());
+ FieldObjectMap fieldMap = new FieldObjectMap();
+ fieldMap.get = PojoUtils.createGetter(inputClass, f.getName(), c);
+ try {
+ fieldMap.set = PojoUtils.createSetter(outputClass, f.getName(), c);
+ } catch (Throwable e) {
+ throw new RuntimeException("Failed to initialize Output Class for field: " + f.getName(), e);
+ }
+ this.fieldMap.add(fieldMap);
+ }
+ }
+
+ private void populateUpdatesFrmIncludeFields() {
+ if (this.outputClass == null) {
+ logger.debug("Creating output class instance from string: {}", outputClassStr);
+ try {
+ this.outputClass = this.getClass().getClassLoader().loadClass(outputClassStr);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ for (String fName : includeFields) {
+ try {
+ Field f = outputClass.getField(fName);
+ Class c;
+ if(f.getType().isPrimitive()) {
+ c = ClassUtils.primitiveToWrapper(f.getType());
+ } else {
+ c = f.getType();
+ }
+ try {
+ updateSetter.add(PojoUtils.createSetter(outputClass, f.getName(), c));
+ } catch (Throwable e) {
+ throw new RuntimeException("Failed to initialize Output Class for field: " + f.getName(), e);
+ }
+ } catch (NoSuchFieldException e) {
+ throw new RuntimeException("Cannot find field '" + fName + "' in output class", e);
+ }
+ }
+ }
+
+ public String getOutputClassStr()
+ {
+ return outputClassStr;
+ }
+
+ public void setOutputClassStr(String outputClassStr)
+ {
+ this.outputClassStr = outputClassStr;
+ }
+
+ @Override protected void processTuple(Object tuple)
+ {
+ if (inputClass == null) {
+ inputClass = tuple.getClass();
+ populateGettersFrmLookup();
+ populateGettersFrmInput();
+ }
+ super.processTuple(tuple);
+ }
+
+ private class FieldObjectMap
+ {
+ public Getter get;
+ public Setter set;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
new file mode 100644
index 0000000..3357704
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
@@ -0,0 +1,38 @@
+/**
+ * Copyright (c) 2015 DataTorrent, Inc.
+ * All rights reserved.
+ */
+package com.datatorrent.contrib.enrichment;
+
+import java.util.List;
+import java.util.Map;
+/**
+ * @since 3.1.0
+ */
+
+public abstract class ReadOnlyBackup implements EnrichmentBackup
+{
+ protected transient List<String> includeFields;
+ protected transient List<String> lookupFields;
+
+ @Override public void put(Object key, Object value)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void putAll(Map<Object, Object> m)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void remove(Object key)
+ {
+ throw new RuntimeException("Not supported operation");
+ }
+
+ @Override public void setFields(List<String> lookupFields,List<String> includeFields)
+ {
+ this.includeFields = includeFields;
+ this.lookupFields = lookupFields;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
new file mode 100644
index 0000000..7d5b4cd
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
@@ -0,0 +1 @@
+package com.datatorrent.contrib.enrichment;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
new file mode 100644
index 0000000..308aa82
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
@@ -0,0 +1,95 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class BeanEnrichmentOperatorTest extends JDBCLoaderTest
+{
+ public class Order {
+ public int OID;
+ public int ID;
+ public double amount;
+
+ public Order(int oid, int id, double amount) {
+ this.OID = oid;
+ this.ID = id;
+ this.amount = amount;
+ }
+ public int getOID()
+ {
+ return OID;
+ }
+
+ public void setOID(int OID)
+ {
+ this.OID = OID;
+ }
+
+ public int getID()
+ {
+ return ID;
+ }
+
+ public void setID(int ID)
+ {
+ this.ID = ID;
+ }
+
+ public double getAmount()
+ {
+ return amount;
+ }
+
+ public void setAmount(double amount)
+ {
+ this.amount = amount;
+ }
+ }
+
+
+ @Test
+ public void includeSelectedKeys()
+ {
+ POJOEnrichmentOperator oper = new POJOEnrichmentOperator();
+ oper.setStore(testMeta.dbloader);
+ oper.setLookupFieldsStr("ID");
+ oper.setIncludeFieldsStr("NAME,AGE,ADDRESS");
+ oper.outputClass = EmployeeOrder.class;
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ oper.beginWindow(1);
+ Order tuple = new Order(3, 4, 700);
+ oper.input.process(tuple);
+ oper.endWindow();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Ouput Tuple: ", "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=0.0}", sink.collectedTuples.get(0).toString());
+ }
+ @Test
+ public void includeAllKeys()
+ {
+ POJOEnrichmentOperator oper = new POJOEnrichmentOperator();
+ oper.setStore(testMeta.dbloader);
+ oper.setLookupFieldsStr("ID");
+ oper.outputClass = EmployeeOrder.class;
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+
+ oper.beginWindow(1);
+ Order tuple = new Order(3, 4, 700);
+ oper.input.process(tuple);
+ oper.endWindow();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Ouput Tuple: ", "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=65000.0}", sink.collectedTuples.get(0).toString());
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
new file mode 100644
index 0000000..00c9d82
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
@@ -0,0 +1,95 @@
+package com.datatorrent.contrib.enrichment;
+
+// This class is needed for Bean Enrichment Operator testing
+public class EmployeeOrder {
+ public int OID;
+ public int ID;
+ public double amount;
+ public String NAME;
+ public int AGE;
+ public String ADDRESS;
+ public double SALARY;
+
+ public int getOID()
+ {
+ return OID;
+ }
+
+ public void setOID(int OID)
+ {
+ this.OID = OID;
+ }
+
+ public int getID()
+ {
+ return ID;
+ }
+
+ public void setID(int ID)
+ {
+ this.ID = ID;
+ }
+
+ public int getAGE()
+ {
+ return AGE;
+ }
+
+ public void setAGE(int AGE)
+ {
+ this.AGE = AGE;
+ }
+
+ public String getNAME()
+ {
+ return NAME;
+ }
+
+ public void setNAME(String NAME)
+ {
+ this.NAME = NAME;
+ }
+
+ public double getAmount()
+ {
+ return amount;
+ }
+
+ public void setAmount(double amount)
+ {
+ this.amount = amount;
+ }
+
+ public String getADDRESS()
+ {
+ return ADDRESS;
+ }
+
+ public void setADDRESS(String ADDRESS)
+ {
+ this.ADDRESS = ADDRESS;
+ }
+
+ public double getSALARY()
+ {
+ return SALARY;
+ }
+
+ public void setSALARY(double SALARY)
+ {
+ this.SALARY = SALARY;
+ }
+
+ @Override public String toString()
+ {
+ return "{" +
+ "OID=" + OID +
+ ", ID=" + ID +
+ ", amount=" + amount +
+ ", NAME='" + NAME + '\'' +
+ ", AGE=" + AGE +
+ ", ADDRESS='" + ADDRESS.trim() + '\'' +
+ ", SALARY=" + SALARY +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
new file mode 100644
index 0000000..934d73b
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
@@ -0,0 +1,75 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import com.esotericsoftware.kryo.Kryo;
+import com.google.common.collect.Maps;
+import org.apache.commons.io.FileUtils;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.util.Map;
+
+
+public class FileEnrichmentTest
+{
+
+ @Rule public final TestUtils.TestInfo testInfo = new TestUtils.TestInfo();
+
+ @Test public void testEnrichmentOperator() throws IOException, InterruptedException
+ {
+ URL origUrl = this.getClass().getResource("/productmapping.txt");
+
+ URL fileUrl = new URL(this.getClass().getResource("/").toString() + "productmapping1.txt");
+ FileUtils.deleteQuietly(new File(fileUrl.getPath()));
+ FileUtils.copyFile(new File(origUrl.getPath()), new File(fileUrl.getPath()));
+
+ MapEnrichmentOperator oper = new MapEnrichmentOperator();
+ FSLoader store = new FSLoader();
+ store.setFileName(fileUrl.toString());
+ oper.setLookupFieldsStr("productId");
+ oper.setStore(store);
+
+ oper.setup(null);
+
+ /* File contains 6 entries, but operator one entry is duplicate,
+ * so cache should contains only 5 entries after scanning input file.
+ */
+ //Assert.assertEquals("Number of mappings ", 7, oper.cache.size());
+
+ CollectorTestSink<Map<String, Object>> sink = new CollectorTestSink<Map<String, Object>>();
+ @SuppressWarnings({ "unchecked", "rawtypes" }) CollectorTestSink<Object> tmp = (CollectorTestSink) sink;
+ oper.output.setSink(tmp);
+
+ oper.beginWindow(0);
+ Map<String, Object> tuple = Maps.newHashMap();
+ tuple.put("productId", 3);
+ tuple.put("channelId", 4);
+ tuple.put("amount", 10.0);
+
+ Kryo kryo = new Kryo();
+ oper.input.process(kryo.copy(tuple));
+
+ oper.endWindow();
+
+ /* Number of tuple, emitted */
+ Assert.assertEquals("Number of tuple emitted ", 1, sink.collectedTuples.size());
+ Map<String, Object> emitted = sink.collectedTuples.iterator().next();
+
+ /* The fields present in original event is kept as it is */
+ Assert.assertEquals("Number of fields in emitted tuple", 4, emitted.size());
+ Assert.assertEquals("value of productId is 3", tuple.get("productId"), emitted.get("productId"));
+ Assert.assertEquals("value of channelId is 4", tuple.get("channelId"), emitted.get("channelId"));
+ Assert.assertEquals("value of amount is 10.0", tuple.get("amount"), emitted.get("amount"));
+
+ /* Check if productCategory is added to the event */
+ Assert.assertEquals("productCategory is part of tuple", true, emitted.containsKey("productCategory"));
+ Assert.assertEquals("value of product category is 1", 5, emitted.get("productCategory"));
+
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
new file mode 100644
index 0000000..07be982
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
@@ -0,0 +1,162 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.netlet.util.DTThrowable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.MasterNotRunningException;
+import org.apache.hadoop.hbase.ZooKeeperConnectionException;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.slf4j.LoggerFactory;
+
+public class HBaseLoaderTest
+{
+ static final org.slf4j.Logger logger = LoggerFactory.getLogger(HBaseLoaderTest.class);
+
+ public static class TestMeta extends TestWatcher
+ {
+
+ HBaseLoader dbloader;
+ @Override
+ protected void starting(Description description)
+ {
+ try {
+ dbloader = new HBaseLoader();
+ Configuration conf = HBaseConfiguration.create();
+ conf.addResource(new Path("file:///home/chaitanya/hbase-site.xml"));
+
+ dbloader.setConfiguration(conf);
+ dbloader.setZookeeperQuorum("localhost");
+ dbloader.setZookeeperClientPort(2181);
+
+ dbloader.setTableName("EMPLOYEE");
+
+ dbloader.connect();
+ createTable();
+ insertRecordsInTable();
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ private void createTable()
+ {
+ try {
+ String[] familys = { "personal", "professional" };
+ HBaseAdmin admin = new HBaseAdmin(dbloader.getConfiguration());
+ HTableDescriptor tableDesc = new HTableDescriptor(dbloader.getTableName());
+ for (int i = 0; i < familys.length; i++) {
+ tableDesc.addFamily(new HColumnDescriptor(familys[i]));
+ }
+ admin.createTable(tableDesc);
+
+ logger.debug("Table created successfully...");
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ public void addRecord(String rowKey, String family, String qualifier, String value) throws Exception {
+ try {
+ HTable table = new HTable(dbloader.getConfiguration(), dbloader.getTableName());
+ Put put = new Put(Bytes.toBytes(rowKey));
+ put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes
+ .toBytes(value));
+ table.put(put);
+ } catch (IOException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+ private void insertRecordsInTable()
+ {
+ try {
+ addRecord("row1", "personal", "name", "raju");
+ addRecord("row1", "personal", "city", "hyderabad");
+ addRecord("row1", "professional", "designation", "manager");
+ addRecord("row1", "professional", "Salary", "50000");
+
+ addRecord("row2", "personal", "name", "ravi");
+ addRecord("row2", "personal", "city", "Chennai");
+ addRecord("row2", "professional", "designation", "SE");
+ addRecord("row2", "professional", "Salary", "30000");
+
+ addRecord("row3", "personal", "name", "rajesh");
+ addRecord("row3", "personal", "city", "Delhi");
+ addRecord("row3", "professional", "designation", "E");
+ addRecord("row3", "professional", "Salary", "10000");
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+
+ }
+
+ private void cleanTable()
+ {
+ String sql = "delete from " + dbloader.getTableName();
+ try {
+ HBaseAdmin admin = new HBaseAdmin(dbloader.getConfiguration());
+ admin.disableTable(dbloader.getTableName());
+ admin.deleteTable(dbloader.getTableName());
+ } catch (MasterNotRunningException e) {
+ e.printStackTrace();
+ } catch (ZooKeeperConnectionException e) {
+ e.printStackTrace();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ protected void finished(Description description)
+ {
+ cleanTable();
+ }
+ }
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ @Test
+ public void testHBaseLookup() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ ArrayList<String> includeKeys = new ArrayList<String>();
+ includeKeys.add("city");
+ includeKeys.add("Salary");
+ ArrayList<String> lookupKeys = new ArrayList<String>();
+ lookupKeys.add("ID");
+ testMeta.dbloader.setFields(lookupKeys, includeKeys);
+
+ String includeFamilyStr = "personal, professional";
+ testMeta.dbloader.setIncludeFamilyStr(includeFamilyStr);
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add("row2");
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("CITY", "Chennai", columnInfo.get(0).toString().trim());
+ Assert.assertEquals("Salary", 30000, columnInfo.get(1));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
new file mode 100644
index 0000000..72cfb88
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
@@ -0,0 +1,179 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.netlet.util.DTThrowable;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.LoggerFactory;
+
+public class JDBCLoaderTest
+{
+ static final org.slf4j.Logger logger = LoggerFactory.getLogger(JDBCLoaderTest.class);
+
+ public static class TestMeta extends TestWatcher
+ {
+ JDBCLoader dbloader;
+ @Override
+ protected void starting(Description description)
+ {
+ try {
+ dbloader = new JDBCLoader();
+ dbloader.setDatabaseDriver("org.hsqldb.jdbcDriver");
+ dbloader.setDatabaseUrl("jdbc:hsqldb:mem:test;sql.syntax_mys=true");
+ dbloader.setTableName("COMPANY");
+
+ dbloader.connect();
+ createTable();
+ insertRecordsInTable();
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ private void createTable()
+ {
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+
+ String createTable = "CREATE TABLE IF NOT EXISTS " + dbloader.getTableName() +
+ "(ID INT PRIMARY KEY NOT NULL," +
+ " NAME TEXT NOT NULL, " +
+ " AGE INT NOT NULL, " +
+ " ADDRESS CHAR(50), " +
+ " SALARY REAL)";
+ stmt.executeUpdate(createTable);
+
+ logger.debug("Table created successfully...");
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ private void insertRecordsInTable()
+ {
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+ String tbName = dbloader.getTableName();
+
+ String sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
+ "VALUES (1, 'Paul', 32, 'California', 20000.00 );";
+ stmt.executeUpdate(sql);
+
+ sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
+ "VALUES (2, 'Allen', 25, 'Texas', 15000.00 );";
+ stmt.executeUpdate(sql);
+
+ sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
+ "VALUES (3, 'Teddy', 23, 'Norway', 20000.00 );";
+ stmt.executeUpdate(sql);
+
+ sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
+ "VALUES (4, 'Mark', 25, 'Rich-Mond', 65000.00 );";
+ stmt.executeUpdate(sql);
+ }
+ catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+
+ }
+
+ private void cleanTable()
+ {
+ String sql = "delete from " + dbloader.tableName;
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+ stmt.executeUpdate(sql);
+ logger.debug("Table deleted successfully...");
+ } catch (SQLException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ protected void finished(Description description)
+ {
+ cleanTable();
+ }
+ }
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ @Test
+ public void testMysqlDBLookup() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ ArrayList<String> lookupKeys = new ArrayList<String>();
+ lookupKeys.add("ID");
+ ArrayList<String> includeKeys = new ArrayList<String>();
+ includeKeys.add("NAME");
+ includeKeys.add("AGE");
+ includeKeys.add("ADDRESS");
+ testMeta.dbloader.setFields(lookupKeys, includeKeys);
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add("4");
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("NAME", "Mark", columnInfo.get(0).toString().trim());
+ Assert.assertEquals("AGE", 25, columnInfo.get(1));
+ Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(2).toString().trim());
+ }
+
+ @Test
+ public void testMysqlDBLookupIncludeAllKeys() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ ArrayList<String> lookupKeys = new ArrayList<String>();
+ lookupKeys.add("ID");
+ ArrayList<String> includeKeys = new ArrayList<String>();
+ testMeta.dbloader.setFields(lookupKeys, includeKeys);
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add("4");
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("ID", 4, columnInfo.get(0));
+ Assert.assertEquals("NAME", "Mark", columnInfo.get(1).toString().trim());
+ Assert.assertEquals("AGE", 25, columnInfo.get(2));
+ Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(3).toString().trim());
+ Assert.assertEquals("SALARY", 65000.0, columnInfo.get(4));
+ }
+
+ @Test
+ public void testMysqlDBQuery() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ testMeta.dbloader.setQueryStmt("Select id, name from " + testMeta.dbloader.getTableName() + " where AGE = ? and ADDRESS = ?");
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add("25");
+ keys.add("Texas");
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("ID", 2, columnInfo.get(0));
+ Assert.assertEquals("NAME", "Allen", columnInfo.get(1).toString().trim());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/528e7aca/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
new file mode 100644
index 0000000..845fe80
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
@@ -0,0 +1,152 @@
+package com.datatorrent.contrib.enrichment;
+
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.collections.CollectionUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MapEnrichmentOperatorTest
+{
+ @Test
+ public void includeAllKeys()
+ {
+ MapEnrichmentOperator oper = new MapEnrichmentOperator();
+ oper.setStore(new MemoryStore());
+ oper.setLookupFieldsStr("In1");
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ Map<String, Object> inMap = Maps.newHashMap();
+ inMap.put("In1", "Value1");
+ inMap.put("In2", "Value2");
+
+ oper.beginWindow(1);
+ oper.input.process(inMap);
+ oper.endWindow();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, C=Val_C, In2=Value2, In1=Value3}", sink.collectedTuples.get(0).toString());
+ }
+
+ @Test
+ public void includeSelectedKeys()
+ {
+ MapEnrichmentOperator oper = new MapEnrichmentOperator();
+ oper.setStore(new MemoryStore());
+ oper.setLookupFieldsStr("In1");
+ oper.setIncludeFieldsStr("A,B");
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ Map<String, Object> inMap = Maps.newHashMap();
+ inMap.put("In1", "Value1");
+ inMap.put("In2", "Value2");
+
+ oper.beginWindow(1);
+ oper.input.process(inMap);
+ oper.endWindow();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, In2=Value2, In1=Value1}", sink.collectedTuples.get(0).toString());
+ }
+
+ private static class MemoryStore implements EnrichmentBackup
+ {
+ static Map<String, Map> returnData = Maps.newHashMap();
+ private List<String> includeFields;
+ static {
+ Map<String, String> map = Maps.newHashMap();
+ map.put("A", "Val_A");
+ map.put("B", "Val_B");
+ map.put("C", "Val_C");
+ map.put("In1", "Value3");
+ returnData.put("Value1", map);
+
+ map = Maps.newHashMap();
+ map.put("A", "Val_A_1");
+ map.put("B", "Val_B_1");
+ map.put("C", "Val_C");
+ map.put("In1", "Value3");
+ returnData.put("Value2", map);
+ }
+
+ @Override public Map<Object, Object> loadInitialData()
+ {
+ return null;
+ }
+
+ @Override public Object get(Object key)
+ {
+ List<String> keyList = (List<String>)key;
+ Map<String, String> keyValue = returnData.get(keyList.get(0));
+ ArrayList<Object> lst = new ArrayList<Object>();
+ if(CollectionUtils.isEmpty(includeFields)) {
+ if(includeFields == null)
+ includeFields = new ArrayList<String>();
+ for (Map.Entry<String, String> e : keyValue.entrySet()) {
+ includeFields.add(e.getKey());
+ }
+ }
+ for(String field : includeFields) {
+ lst.add(keyValue.get(field));
+ }
+ return lst;
+ }
+
+ @Override public List<Object> getAll(List<Object> keys)
+ {
+ return null;
+ }
+
+ @Override public void put(Object key, Object value)
+ {
+
+ }
+
+ @Override public void putAll(Map<Object, Object> m)
+ {
+
+ }
+
+ @Override public void remove(Object key)
+ {
+
+ }
+
+ @Override public void connect() throws IOException
+ {
+
+ }
+
+ @Override public void disconnect() throws IOException
+ {
+
+ }
+
+ @Override public boolean isConnected()
+ {
+ return false;
+ }
+
+ @Override public void setFields(List<String> lookupFields, List<String> includeFields)
+ {
+ this.includeFields = includeFields;
+
+ }
+
+ @Override
+ public boolean needRefresh() {
+ return false;
+ }
+ }
+}
[3/4] incubator-apex-malhar git commit: APEXMALHAR-2023 Adding
Enrichment Operator to Malhar
Posted by sa...@apache.org.
APEXMALHAR-2023 Adding Enrichment Operator to Malhar
Added 2 operators POJOEnricher and MapEnricher which enriches the given POJO or map as configured.
The operators are marked evolving.
Test cases added for both operators.
2 Backend loaders are added for File and JDBC.
Hbase loader removed as there are some things that needs to be discussed. Postponed the implementation for after discussion.
Added a new SupportType for OBJECT to FieldInfo.
Added static method in FieldInfo for conversion to Class -> SupportedType.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/9600eddd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/9600eddd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/9600eddd
Branch: refs/heads/master
Commit: 9600eddd8ee4fa010cbb4799da74e89d184daa40
Parents: 528e7ac
Author: chinmaykolhatkar <ch...@datatorrent.com>
Authored: Tue Mar 29 14:46:54 2016 +0530
Committer: chinmaykolhatkar <ch...@datatorrent.com>
Committed: Fri Apr 29 22:59:48 2016 +0530
----------------------------------------------------------------------
.../contrib/enrich/AbstractEnricher.java | 321 +++++++++++++++++++
.../contrib/enrich/BackendLoader.java | 42 +++
.../datatorrent/contrib/enrich/FSLoader.java | 184 +++++++++++
.../datatorrent/contrib/enrich/JDBCLoader.java | 201 ++++++++++++
.../datatorrent/contrib/enrich/MapEnricher.java | 138 ++++++++
.../contrib/enrich/NullValuesCacheManager.java | 60 ++++
.../contrib/enrich/POJOEnricher.java | 287 +++++++++++++++++
.../contrib/enrich/ReadOnlyBackup.java | 61 ++++
.../enrichment/AbstractEnrichmentOperator.java | 170 ----------
.../contrib/enrichment/EnrichmentBackup.java | 18 --
.../contrib/enrichment/FSLoader.java | 146 ---------
.../contrib/enrichment/HBaseLoader.java | 128 --------
.../contrib/enrichment/JDBCLoader.java | 158 ---------
.../enrichment/MapEnrichmentOperator.java | 57 ----
.../enrichment/NullValuesCacheManager.java | 43 ---
.../enrichment/POJOEnrichmentOperator.java | 185 -----------
.../contrib/enrichment/ReadOnlyBackup.java | 38 ---
.../contrib/enrichment/package-info.java | 1 -
.../contrib/enrich/EmployeeOrder.java | 114 +++++++
.../contrib/enrich/FileEnrichmentTest.java | 103 ++++++
.../contrib/enrich/JDBCLoaderTest.java | 209 ++++++++++++
.../contrib/enrich/MapEnricherTest.java | 251 +++++++++++++++
.../com/datatorrent/contrib/enrich/Order.java | 71 ++++
.../contrib/enrich/POJOEnricherTest.java | 232 ++++++++++++++
.../enrichment/BeanEnrichmentOperatorTest.java | 95 ------
.../contrib/enrichment/EmployeeOrder.java | 95 ------
.../contrib/enrichment/FileEnrichmentTest.java | 75 -----
.../contrib/enrichment/HBaseLoaderTest.java | 162 ----------
.../contrib/enrichment/JDBCLoaderTest.java | 179 -----------
.../enrichment/MapEnrichmentOperatorTest.java | 152 ---------
contrib/src/test/resources/productmapping.txt | 100 ++++++
.../com/datatorrent/lib/util/FieldInfo.java | 15 +-
32 files changed, 2388 insertions(+), 1703 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java
new file mode 100644
index 0000000..cdefddf
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java
@@ -0,0 +1,321 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.esotericsoftware.kryo.NotNull;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.db.cache.CacheManager;
+import com.datatorrent.lib.db.cache.CacheStore;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.FieldInfo.SupportType;
+
+/**
+ * Base class for Enrichment Operator. Subclasses should provide implementation to getKey and convert.
+ * The operator receives a tuple and emits enriched tuple based on includeFields and lookupFields. <br/>
+ * <p>
+ * Properties:<br>
+ * <b>lookupFields</b>: List of comma separated keys for quick searching. Ex: Field1,Field2,Field3<br>
+ * <b>includeFields</b>: List of comma separated fields to be replaced/added to the input tuple. Ex: Field1,Field2,Field3<br>
+ * <b>store</b>: Specify the type of loader for looking data<br>
+ * <br>
+ *
+ * @param <INPUT> Type of tuples which are received by this operator</T>
+ * @param <OUTPUT> Type of tuples which are emitted by this operator</T>
+ * @displayName Abstract Enrichment Operator
+ * @tags Enrichment
+ */
+@InterfaceStability.Evolving
+public abstract class AbstractEnricher<INPUT, OUTPUT> extends BaseOperator implements Operator.ActivationListener
+{
+ /**
+ * Mandatory parameters for Enricher
+ */
+ @NotNull
+ protected List<String> lookupFields;
+ @NotNull
+ protected List<String> includeFields;
+ @NotNull
+ private BackendLoader store;
+
+ /**
+ * Optional parameters for enricher.
+ */
+ private int cacheExpirationInterval = 1 * 60 * 60 * 1000; // 1 hour
+ private int cacheCleanupInterval = 1 * 60 * 60 * 1000; // 1 hour
+ private int cacheSize = 1024; // 1024 records
+
+ /**
+ * Helper variables.
+ */
+ private transient CacheManager cacheManager;
+ protected transient List<FieldInfo> lookupFieldInfo = new ArrayList<>();
+ protected transient List<FieldInfo> includeFieldInfo = new ArrayList<>();
+
+ /**
+ * This method needs to be called by implementing class for processing a tuple for enrichment.
+ * The method will take the tuple through following stages:
+ * <ol>
+ * <li>Call {@link #getKey(Object)} to retrieve key fields for lookup</li>
+ * <li>Using key fields call cache manager to retrieve for any key that is cached already</li>
+ * <li>If not found in cache, it'll do a lookup in configured backend store</li>
+ * <li>The retrieved fields will be passed to {@link #convert(Object, Object)} method to create the final object</li>
+ * <li>Finally {@link #emitEnrichedTuple(Object)} is called for emitting the tuple</li>
+ * </ol>
+ *
+ * @param tuple Input tuple that needs to get processed for enrichment.
+ */
+ protected void enrichTuple(INPUT tuple)
+ {
+ Object key = getKey(tuple);
+ if (key != null) {
+ Object result = cacheManager.get(key);
+ OUTPUT out = convert(tuple, result);
+ if (out != null) {
+ emitEnrichedTuple(out);
+ }
+ }
+ }
+
+ /**
+ * The method should be implemented by concrete class which returns an ArrayList<Object> containing all the fields
+ * which forms key part of lookup.
+ * The order of field values should be same as the one set in {@link #lookupFields} variable.
+ *
+ * @param tuple Input tuple from which fields values for key needs to be fetched.
+ * @return Should return ArrayList<Object> which has fields values forming keys in same order as {@link #lookupFields}
+ */
+ protected abstract Object getKey(INPUT tuple);
+
+ /**
+ * The method should be implemented by concrete class.
+ * This method is expected to take input tuple and an externally fetched object containing fields to be enriched, and
+ * return an enriched tuple which is ready to be emitted.
+ *
+ * @param in Input tuple which needs to be enriched.
+ * @param cached ArrayList<Object> containing missing data retrieved from external sources.
+ * @return Enriched tuple of type OUTPUT
+ */
+ protected abstract OUTPUT convert(INPUT in, Object cached);
+
+ /**
+ * This method should be implemented by concrete class.
+ * The method is expected to emit tuple of type OUTPUT
+ *
+ * @param tuple Tuple of type OUTPUT that should be emitted.
+ */
+ protected abstract void emitEnrichedTuple(OUTPUT tuple);
+
+ /**
+ * This method should be implemented by concrete method.
+ * The method should return Class type of field for given fieldName from output tuple.
+ *
+ * @param fieldName Field name for which field type needs to be identified
+ * @return Class type for given field.
+ */
+ protected abstract Class<?> getIncludeFieldType(String fieldName);
+
+ /**
+ * This method should be implemented by concrete method.
+ * The method should return Class type of field for given fieldName from input tuple.
+ *
+ * @param fieldName Field name for which field type needs to be identified
+ * @return Class type for given field.
+ */
+ protected abstract Class<?> getLookupFieldType(String fieldName);
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+
+ cacheManager = new NullValuesCacheManager();
+ CacheStore primaryCache = new CacheStore();
+
+ // set expiration to one day.
+ primaryCache.setEntryExpiryDurationInMillis(cacheExpirationInterval);
+ primaryCache.setCacheCleanupInMillis(cacheCleanupInterval);
+ primaryCache.setEntryExpiryStrategy(CacheStore.ExpiryType.EXPIRE_AFTER_WRITE);
+ primaryCache.setMaxCacheSize(cacheSize);
+
+ cacheManager.setPrimary(primaryCache);
+ cacheManager.setBackup(store);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ for (String s : lookupFields) {
+ lookupFieldInfo.add(new FieldInfo(s, s, SupportType.getFromJavaType(getLookupFieldType(s))));
+ }
+
+ if (includeFields != null) {
+ for (String s : includeFields) {
+ includeFieldInfo.add(new FieldInfo(s, s, SupportType.getFromJavaType(getIncludeFieldType(s))));
+ }
+ }
+
+ store.setFieldInfo(lookupFieldInfo, includeFieldInfo);
+
+ try {
+ cacheManager.initialize();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to initialize primary cache", e);
+ }
+ }
+
+ @Override
+ public void deactivate()
+ {
+ }
+
+ /**
+ * Returns a list of fields which are used for lookup.
+ *
+ * @return List of fields
+ */
+ public List<String> getLookupFields()
+ {
+ return lookupFields;
+ }
+
+ /**
+ * Set fields on which lookup needs to happen in external store.
+ * This is a mandatory parameter.
+ *
+ * @param lookupFields List of fields on which lookup happens.
+ */
+ public void setLookupFields(List<String> lookupFields)
+ {
+ this.lookupFields = lookupFields;
+ }
+
+ /**
+ * Returns a list of fields using which tuple is enriched
+ *
+ * @return List of fields.
+ */
+ public List<String> getIncludeFields()
+ {
+ return includeFields;
+ }
+
+ /**
+ * Sets list of fields to be fetched from external store for enriching the tuple.
+ * This is a mandatory parameter.
+ *
+ * @param includeFields List of fields.
+ */
+ public void setIncludeFields(List<String> includeFields)
+ {
+ this.includeFields = includeFields;
+ }
+
+ /**
+ * Returns the backend store which will enrich the tuple.
+ *
+ * @return Object of type {@link BackendLoader}
+ */
+ public BackendLoader getStore()
+ {
+ return store;
+ }
+
+ /**
+ * Sets backend store which will enrich the tuple.
+ * This is a mandatory parameter.
+ *
+ * @param store Object of type {@link BackendLoader}
+ */
+ public void setStore(BackendLoader store)
+ {
+ this.store = store;
+ }
+
+ /**
+ * Returns cache entry expiration interval in ms.
+ * This is an optional parameter.
+ *
+ * @return Cache entry expiration interval in ms
+ */
+ public int getCacheExpirationInterval()
+ {
+ return cacheExpirationInterval;
+ }
+
+ /**
+ * Sets cache entry expiration interval in ms.
+ * This is an optional parameter.
+ *
+ * @param cacheExpirationInterval Cache entry expiration interval in ms
+ */
+ public void setCacheExpirationInterval(int cacheExpirationInterval)
+ {
+ this.cacheExpirationInterval = cacheExpirationInterval;
+ }
+
+ /**
+ * Returns cache cleanup interval in ms. After this interval, cache cleanup operation will be performed.
+ * This is an optional parameter.
+ *
+ * @return cache cleanup interval in ms.
+ */
+ public int getCacheCleanupInterval()
+ {
+ return cacheCleanupInterval;
+ }
+
+ /**
+ * Set Cache cleanup interval in ms. After this interval, cache cleanup operation will be performed.
+ * This is an optional parameter.
+ *
+ * @param cacheCleanupInterval cache cleanup interval in ms.
+ */
+ public void setCacheCleanupInterval(int cacheCleanupInterval)
+ {
+ this.cacheCleanupInterval = cacheCleanupInterval;
+ }
+
+ /**
+ * Get size (number of entries) of cache.
+ *
+ * @return Number of entries allowed in cache.
+ */
+ public int getCacheSize()
+ {
+ return cacheSize;
+ }
+
+ /**
+ * Set size (number of entries) of cache.
+ *
+ * @param cacheSize Number of entries allowed in cache.
+ */
+ public void setCacheSize(int cacheSize)
+ {
+ this.cacheSize = cacheSize;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/BackendLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/BackendLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/BackendLoader.java
new file mode 100644
index 0000000..9570329
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/BackendLoader.java
@@ -0,0 +1,42 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.util.List;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.lib.db.cache.CacheManager;
+import com.datatorrent.lib.util.FieldInfo;
+
+/**
+ * Interface for store to be used in enrichment
+ */
+@InterfaceStability.Evolving
+public interface BackendLoader extends CacheManager.Backup
+{
+ /**
+ * Set {@link FieldInfo} for lookup fields and also include fields.
+ * Calling this method is mandatory for correct functioning of backend loader.
+ *
+ * @param lookupFieldInfo List of {@link FieldInfo} that will be used as key in lookup.
+ * @param includeFieldInfo List of {@link FieldInfo} that will be retrieved from store.
+ */
+ void setFieldInfo(List<FieldInfo> lookupFieldInfo, List<FieldInfo> includeFieldInfo);
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java
new file mode 100644
index 0000000..71d3dce
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java
@@ -0,0 +1,184 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.codehaus.jackson.JsonProcessingException;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.ObjectReader;
+import org.codehaus.jackson.type.TypeReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.esotericsoftware.kryo.NotNull;
+import com.google.common.collect.Maps;
+import com.datatorrent.lib.db.cache.CacheManager;
+import com.datatorrent.lib.util.FieldInfo;
+
+
+/**
+ * This implementation of {@link BackendLoader} loads the data from a given file into memory cache and serves queries
+ * from the cache.
+ * When this is set as primaryCache in {@link CacheManager}, CacheManager can call {@link #loadInitialData()}
+ * periodically to reload the file.
+ * <p>
+ * The format of the input file is:
+ * <p>
+ * {"productCategory": 5, "productId": 0}
+ * {"productCategory": 4, "productId": 1}
+ * {"productCategory": 5, "productId": 2}
+ * {"productCategory": 5, "productId": 3}
+ * </p>
+ * Each line in the input file should be a valid json object which represents a record and each key/value pair in that
+ * json object represents the fields/value.
+ * <p>
+ * NOTE: This loader should be used with caution as all the data present in the file is loaded in memory because of
+ * which the memory consumption may go up.
+ */
+@InterfaceStability.Evolving
+public class FSLoader extends ReadOnlyBackup
+{
+ @NotNull
+ private String fileName;
+
+ private transient Path filePath;
+ private transient FileSystem fs;
+ private transient boolean connected;
+
+ private static final ObjectMapper mapper = new ObjectMapper();
+ private static final ObjectReader reader = mapper.reader(new TypeReference<Map<String, Object>>(){});
+ private static final Logger logger = LoggerFactory.getLogger(FSLoader.class);
+
+ public String getFileName()
+ {
+ return fileName;
+ }
+
+ public void setFileName(String fileName)
+ {
+ this.fileName = fileName;
+ }
+
+ @Override
+ public Map<Object, Object> loadInitialData()
+ {
+ Map<Object, Object> result = null;
+ FSDataInputStream in = null;
+ BufferedReader bin = null;
+ try {
+ result = Maps.newHashMap();
+ in = fs.open(filePath);
+ bin = new BufferedReader(new InputStreamReader(in));
+ String line;
+ while ((line = bin.readLine()) != null) {
+ try {
+ Map<String, Object> tuple = reader.readValue(line);
+ result.put(getKey(tuple), getValue(tuple));
+ } catch (JsonProcessingException parseExp) {
+ logger.info("Unable to parse line {}", line);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ if (bin != null) {
+ IOUtils.closeQuietly(bin);
+ }
+ if (in != null) {
+ IOUtils.closeQuietly(in);
+ }
+ try {
+ fs.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ logger.debug("loading initial data {}", result.size());
+ return result;
+ }
+
+ private Object getValue(Map<String, Object> tuple)
+ {
+ ArrayList<Object> includeTuple = new ArrayList<Object>();
+ for (FieldInfo s : includeFieldInfo) {
+ includeTuple.add(tuple.get(s.getColumnName()));
+ }
+ return includeTuple;
+ }
+
+ private Object getKey(Map<String, Object> tuple)
+ {
+ ArrayList<Object> list = new ArrayList<Object>();
+ for (FieldInfo key : lookupFieldInfo) {
+ list.add(tuple.get(key.getColumnName()));
+ }
+ return list;
+ }
+
+ @Override
+ public Object get(Object key)
+ {
+ return null;
+ }
+
+ @Override
+ public List<Object> getAll(List<Object> keys)
+ {
+ return null;
+ }
+
+ @Override
+ public void connect() throws IOException
+ {
+ Configuration conf = new Configuration();
+ this.filePath = new Path(fileName);
+ this.fs = FileSystem.newInstance(filePath.toUri(), conf);
+ if (!fs.isFile(filePath)) {
+ throw new IOException("Provided path " + fileName + " is not a file");
+ }
+ connected = true;
+ }
+
+ @Override
+ public void disconnect() throws IOException
+ {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+
+ @Override
+ public boolean isConnected()
+ {
+ return connected;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/JDBCLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/JDBCLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/JDBCLoader.java
new file mode 100644
index 0000000..d20e87b
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/JDBCLoader.java
@@ -0,0 +1,201 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.collect.Lists;
+import com.datatorrent.lib.db.jdbc.JdbcStore;
+import com.datatorrent.lib.util.FieldInfo;
+
+/**
+ * <p>HBaseLoader extends from {@link JdbcStore} uses JDBC to connect and implements BackendLoaders interface.</p> <br/>
+ * <p>
+ * Properties:<br>
+ * <b>queryStmt</b>: Sql Prepared Statement which needs to be executed<br>
+ * <b>tableName</b>: JDBC table name<br>
+ * <br>
+ */
+@InterfaceStability.Evolving
+public class JDBCLoader extends JdbcStore implements BackendLoader
+{
+ protected String queryStmt;
+
+ protected String tableName;
+
+ protected transient List<FieldInfo> includeFieldInfo;
+ protected transient List<FieldInfo> lookupFieldInfo;
+
+ protected Object getQueryResult(Object key)
+ {
+ try {
+ PreparedStatement getStatement = getConnection().prepareStatement(queryStmt);
+ ArrayList<Object> keys = (ArrayList<Object>)key;
+ for (int i = 0; i < keys.size(); i++) {
+ getStatement.setObject(i + 1, keys.get(i));
+ }
+ return getStatement.executeQuery();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected ArrayList<Object> getDataFrmResult(Object result) throws RuntimeException
+ {
+ try {
+ ResultSet resultSet = (ResultSet)result;
+ if (resultSet.next()) {
+ ResultSetMetaData rsdata = resultSet.getMetaData();
+ // If the includefields is empty, populate it from ResultSetMetaData
+ if (CollectionUtils.isEmpty(includeFieldInfo)) {
+ if (includeFieldInfo == null) {
+ includeFieldInfo = new ArrayList<>();
+ }
+ for (int i = 1; i <= rsdata.getColumnCount(); i++) {
+ String columnName = rsdata.getColumnName(i);
+ // TODO: Take care of type conversion.
+ includeFieldInfo.add(new FieldInfo(columnName, columnName, FieldInfo.SupportType.OBJECT));
+ }
+ }
+
+ ArrayList<Object> res = new ArrayList<Object>();
+ for (FieldInfo f : includeFieldInfo) {
+ res.add(getConvertedData(resultSet.getObject(f.getColumnName()), f));
+ }
+ return res;
+ } else {
+ return null;
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Object getConvertedData(Object object, FieldInfo f)
+ {
+ if (f.getType().getJavaType() == object.getClass()) {
+ return object;
+ } else {
+ logger.warn("Type mismatch seen for field {}, returning as it is", f.getColumnName());
+ return null;
+ }
+ }
+
+ private String generateQueryStmt()
+ {
+ String stmt = "select * from " + tableName + " where ";
+ boolean first = true;
+ for (FieldInfo fieldInfo : lookupFieldInfo) {
+ if (first) {
+ first = false;
+ } else {
+ stmt += " and ";
+ }
+ stmt += fieldInfo.getColumnName() + " = ?";
+ }
+
+ logger.info("generateQueryStmt: {}", stmt);
+ return stmt;
+ }
+
+ public String getQueryStmt()
+ {
+ return queryStmt;
+ }
+
+ /**
+ * Set the sql Prepared Statement if the enrichment mechanism is query based.
+ */
+ public void setQueryStmt(String queryStmt)
+ {
+ this.queryStmt = queryStmt;
+ }
+
+ public String getTableName()
+ {
+ return tableName;
+ }
+
+ /**
+ * Set the table name.
+ */
+ public void setTableName(String tableName)
+ {
+ this.tableName = tableName;
+ }
+
+ @Override
+ public void setFieldInfo(List<FieldInfo> lookupFieldInfo, List<FieldInfo> includeFieldInfo)
+ {
+ this.lookupFieldInfo = lookupFieldInfo;
+ this.includeFieldInfo = includeFieldInfo;
+ if (queryStmt == null) {
+ queryStmt = generateQueryStmt();
+ }
+ }
+
+ @Override
+ public Map<Object, Object> loadInitialData()
+ {
+ return null;
+ }
+
+ @Override
+ public Object get(Object key)
+ {
+ return getDataFrmResult(getQueryResult(key));
+ }
+
+ @Override
+ public List<Object> getAll(List<Object> keys)
+ {
+ List<Object> values = Lists.newArrayList();
+ for (Object key : keys) {
+ values.add(get(key));
+ }
+ return values;
+ }
+
+ @Override
+ public void put(Object key, Object value)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+
+ @Override
+ public void putAll(Map<Object, Object> m)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+
+ @Override
+ public void remove(Object key)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/MapEnricher.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/MapEnricher.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/MapEnricher.java
new file mode 100644
index 0000000..ecf16ba
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/MapEnricher.java
@@ -0,0 +1,138 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.lib.util.FieldInfo;
+
+/**
+ * This class takes a HashMap tuple as input and extracts value of the lookupKey configured
+ * for this operator. It perform a lookup using {@link com.datatorrent.lib.db.cache.CacheManager} to
+ * find a matching entry and adds the result to the original tuple.
+ *
+ * <p>
+ * Example:
+ * Lets say, input tuple is
+ * { amount=10.0, channelId=4, productId=3 }
+ * The tuple is modified as below:
+ * { amount=10.0, channelId=4, productId=3, <b>productCategory=1 </b>}
+ * </p>
+ *
+ * @displayName MapEnricher
+ * @category Database
+ * @tags enrichment, lookup, map
+ */
+@InterfaceStability.Evolving
+public class MapEnricher extends AbstractEnricher<Map<String, Object>, Map<String, Object>>
+{
+ public final transient DefaultInputPort<Map<String, Object>> input = new DefaultInputPort<Map<String, Object>>()
+ {
+ @Override
+ public void process(Map<String, Object> obj)
+ {
+ processTuple(obj);
+ }
+ };
+
+ public final transient DefaultOutputPort<Map<String, Object>> output = new DefaultOutputPort<>();
+
+ protected void processTuple(Map<String, Object> obj)
+ {
+ enrichTuple(obj);
+ }
+
+ @Override
+ protected Object getKey(Map<String, Object> tuple)
+ {
+ ArrayList<Object> keyList = new ArrayList<Object>();
+
+ for (FieldInfo fieldInfo : lookupFieldInfo) {
+ keyList.add(tuple.get(fieldInfo.getColumnName()));
+ }
+
+ return keyList;
+ }
+
+ @Override
+ protected Map<String, Object> convert(Map<String, Object> in, Object cached)
+ {
+ if (cached == null) {
+ return in;
+ }
+
+ ArrayList<Object> newAttributes = (ArrayList<Object>)cached;
+ if (newAttributes != null) {
+ for (int i = 0; i < includeFieldInfo.size(); i++) {
+ in.put(includeFieldInfo.get(i).getColumnName(), newAttributes.get(i));
+ }
+ }
+ return in;
+ }
+
+ @Override
+ protected void emitEnrichedTuple(Map<String, Object> tuple)
+ {
+ output.emit(tuple);
+ }
+
+ @Override
+ protected Class<?> getIncludeFieldType(String fieldName)
+ {
+ return Object.class;
+ }
+
+ @Override
+ protected Class<?> getLookupFieldType(String fieldName)
+ {
+ return Object.class;
+ }
+
+ /**
+ * Set fields on which lookup against which lookup will be performed.
+ * This is a mandatory parameter to set.
+ *
+ * @param lookupFields List of fields on which lookup happens.
+ * @description $[] Field which become part of lookup key
+ */
+ @Override
+ public void setLookupFields(List<String> lookupFields)
+ {
+ super.setLookupFields(lookupFields);
+ }
+
+ /**
+ * Set fields which will enrich the map.
+ * This is a mandatory parameter to set.
+ *
+ * @param includeFields List of fields.
+ * @description $[] Field which are fetched from store
+ */
+ @Override
+ public void setIncludeFields(List<String> includeFields)
+ {
+ super.setIncludeFields(includeFields);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/NullValuesCacheManager.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/NullValuesCacheManager.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/NullValuesCacheManager.java
new file mode 100644
index 0000000..2cf7326
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/NullValuesCacheManager.java
@@ -0,0 +1,60 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.lib.db.cache.CacheManager;
+
+/**
+ * Null Values Cache Manager. Using this NULL entries can be specified explicitly.
+ */
+@InterfaceStability.Evolving
+public class NullValuesCacheManager extends CacheManager
+{
+
+ private static final NullObject NULL = new NullObject();
+
+ @Override
+ public Object get(Object key)
+ {
+ Object primaryVal = primary.get(key);
+ if (primaryVal != null) {
+ if (primaryVal == NULL) {
+ return null;
+ }
+
+ return primaryVal;
+ }
+
+ Object backupVal = backup.get(key);
+ if (backupVal != null) {
+ primary.put(key, backupVal);
+ } else {
+ primary.put(key, NULL);
+ }
+
+ return backupVal;
+ }
+
+ private static class NullObject
+ {
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/POJOEnricher.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/POJOEnricher.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/POJOEnricher.java
new file mode 100644
index 0000000..782fbc5
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/POJOEnricher.java
@@ -0,0 +1,287 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.PojoUtils;
+
+
+/**
+ * This class takes a POJO as input and extracts the value of the lookupKey configured
+ * for this operator. It perform a lookup using {@link com.datatorrent.lib.db.cache.CacheManager} to
+ * find a matching entry and adds the result to the original tuple.
+ *
+ * <p>
+ * Properties:<br>
+ * <b>inputClass</b>: Class to be loaded for the incoming data type<br>
+ * <b>outputClass</b>: Class to be loaded for the emitted data type<br>
+ * <br>
+ * <p>
+ *
+ * <p>
+ * Example:
+ * Lets say, input tuple is
+ * { amount=10.0, channelId=4, productId=3 }
+ * The tuple is modified as below:
+ * { amount=10.0, channelId=4, productId=3, <b>productCategory=1 </b>}
+ * </p>
+ *
+ * @displayName POJOEnricher
+ * @category Database
+ * @tags enrichment, enricher, pojo, schema, lookup
+ */
+@InterfaceStability.Evolving
+public class POJOEnricher extends AbstractEnricher<Object, Object>
+{
+ private static final Logger logger = LoggerFactory.getLogger(POJOEnricher.class);
+
+ /**
+ * Helper fields
+ */
+ protected Class<?> inputClass;
+ protected Class<?> outputClass;
+ private transient Map<PojoUtils.Getter, PojoUtils.Setter> fieldMap = new HashMap<>();
+ private transient List<PojoUtils.Setter> includeSetters = new ArrayList<>();
+ private transient List<PojoUtils.Getter> lookupGetters = new ArrayList<>();
+
+ /**
+ * AutoMetrics
+ */
+ @AutoMetric
+ private int enrichedTupleCount;
+ @AutoMetric
+ private int errorTupleCount;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ inputClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object object)
+ {
+ processTuple(object);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
+ {
+ @Override
+ public void setup(Context.PortContext context)
+ {
+ outputClass = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+ };
+
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<>();
+
+ protected void processTuple(Object object)
+ {
+ enrichTuple(object);
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ enrichedTupleCount = 0;
+ errorTupleCount = 0;
+ }
+
+ @Override
+ protected Object getKey(Object tuple)
+ {
+ ArrayList<Object> keyList = new ArrayList<>();
+ for (PojoUtils.Getter lookupGetter : lookupGetters) {
+ keyList.add(lookupGetter.get(tuple));
+ }
+ return keyList;
+ }
+
+ @Override
+ protected Object convert(Object in, Object cached)
+ {
+ Object o;
+
+ try {
+ o = outputClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException e) {
+ logger.error("Failed to create new instance of output POJO", e);
+ errorTupleCount++;
+ error.emit(in);
+ return null;
+ }
+
+ try {
+ for (Map.Entry<PojoUtils.Getter, PojoUtils.Setter> entry : fieldMap.entrySet()) {
+ entry.getValue().set(o, entry.getKey().get(in));
+ }
+ } catch (RuntimeException e) {
+ logger.error("Failed to set the property. Continuing with default.", e);
+ errorTupleCount++;
+ error.emit(in);
+ return null;
+ }
+
+ if (cached == null) {
+ return o;
+ }
+
+ ArrayList<Object> includeObjects = (ArrayList<Object>)cached;
+ for (int i = 0; i < includeSetters.size(); i++) {
+ try {
+ includeSetters.get(i).set(o, includeObjects.get(i));
+ } catch (RuntimeException e) {
+ logger.error("Failed to set the property. Continuing with default.", e);
+ errorTupleCount++;
+ error.emit(in);
+ return null;
+ }
+ }
+
+ return o;
+ }
+
+ @Override
+ protected void emitEnrichedTuple(Object tuple)
+ {
+ output.emit(tuple);
+ enrichedTupleCount++;
+ }
+
+ @Override
+ protected Class<?> getIncludeFieldType(String fieldName)
+ {
+ try {
+ return outputClass.getDeclaredField(fieldName).getType();
+ } catch (NoSuchFieldException e) {
+ logger.warn("Failed to find given fieldName, returning object type", e);
+ return Object.class;
+ }
+ }
+
+ @Override
+ protected Class<?> getLookupFieldType(String fieldName)
+ {
+ try {
+ return inputClass.getDeclaredField(fieldName).getType();
+ } catch (NoSuchFieldException e) {
+ logger.warn("Failed to find given fieldName, returning object type", e);
+ return Object.class;
+ }
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private PojoUtils.Setter generateSettersForField(Class<?> klass, String outputFieldName)
+ throws NoSuchFieldException, SecurityException
+ {
+ Field f = klass.getDeclaredField(outputFieldName);
+ Class c = ClassUtils.primitiveToWrapper(f.getType());
+ return PojoUtils.createSetter(klass, outputFieldName, c);
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private PojoUtils.Getter generateGettersForField(Class<?> klass, String inputFieldName)
+ throws NoSuchFieldException, SecurityException
+ {
+ Field f = klass.getDeclaredField(inputFieldName);
+ Class c = ClassUtils.primitiveToWrapper(f.getType());
+ return PojoUtils.createGetter(klass, inputFieldName, c);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ super.activate(context);
+
+ for (Field field : inputClass.getDeclaredFields()) {
+ try {
+ fieldMap.put(generateGettersForField(inputClass, field.getName()),
+ generateSettersForField(outputClass, field.getName()));
+ } catch (NoSuchFieldException e) {
+ throw new RuntimeException("Unable to find field with name " + field.getName() + ", ignoring that field.", e);
+ }
+ }
+
+ for (FieldInfo fieldInfo : this.includeFieldInfo) {
+ try {
+ includeSetters.add(generateSettersForField(outputClass, fieldInfo.getColumnName()));
+ } catch (NoSuchFieldException e) {
+ throw new RuntimeException("Given field name is not present in output POJO", e);
+ }
+ }
+
+ for (FieldInfo fieldInfo : this.lookupFieldInfo) {
+ try {
+ lookupGetters.add(generateGettersForField(inputClass, fieldInfo.getColumnName()));
+ } catch (NoSuchFieldException e) {
+ throw new RuntimeException("Given lookup field is not present in POJO", e);
+ }
+ }
+ }
+
+ /**
+ * Set fields on which lookup against which lookup will be performed.
+ * This is a mandatory parameter to set.
+ *
+ * @param lookupFields List of fields on which lookup happens.
+ * @description $[] Field which become part of lookup key
+ * @useSchema $[] input.fields[].name
+ */
+ @Override
+ public void setLookupFields(List<String> lookupFields)
+ {
+ super.setLookupFields(lookupFields);
+ }
+
+ /**
+ * Set fields which will enrich the POJO.
+ * This is a mandatory parameter to set.
+ *
+ * @param includeFields List of fields.
+ * @description $[] Field which are fetched from store
+ * @useSchema $[] output.fields[].name
+ */
+ @Override
+ public void setIncludeFields(List<String> includeFields)
+ {
+ super.setIncludeFields(includeFields);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrich/ReadOnlyBackup.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrich/ReadOnlyBackup.java b/contrib/src/main/java/com/datatorrent/contrib/enrich/ReadOnlyBackup.java
new file mode 100644
index 0000000..157dbc9
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/enrich/ReadOnlyBackup.java
@@ -0,0 +1,61 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.lib.util.FieldInfo;
+
+/**
+ * ReadOnly abstract implementation of BackendLoader.
+ */
+@InterfaceStability.Evolving
+public abstract class ReadOnlyBackup implements BackendLoader
+{
+ protected transient List<FieldInfo> includeFieldInfo;
+ protected transient List<FieldInfo> lookupFieldInfo;
+
+ @Override
+ public void put(Object key, Object value)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+
+ @Override
+ public void putAll(Map<Object, Object> m)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+
+ @Override
+ public void remove(Object key)
+ {
+ throw new UnsupportedOperationException("Not supported operation");
+ }
+
+ @Override
+ public void setFieldInfo(List<FieldInfo> lookupFieldInfo, List<FieldInfo> includeFieldInfo)
+ {
+ this.includeFieldInfo = includeFieldInfo;
+ this.lookupFieldInfo = lookupFieldInfo;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
deleted file mode 100644
index cbd4d5e..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/AbstractEnrichmentOperator.java
+++ /dev/null
@@ -1,170 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.common.util.BaseOperator;
-import com.datatorrent.api.Context;
-import com.datatorrent.api.DefaultInputPort;
-import com.datatorrent.api.DefaultOutputPort;
-import com.datatorrent.api.annotation.InputPortFieldAnnotation;
-import com.datatorrent.lib.db.cache.CacheManager;
-import com.datatorrent.lib.db.cache.CacheStore;
-import com.datatorrent.lib.db.cache.CacheStore.ExpiryType;
-import com.esotericsoftware.kryo.NotNull;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- * Base class for Enrichment Operator. Subclasses should provide implementation to getKey and convert.
- * The operator receives a tuple and emits enriched tuple based on includeFields and lookupFields. <br/>
- *
- * Properties:<br>
- * <b>lookupFieldsStr</b>: List of comma separated keys for quick searching. Ex: Field1,Field2,Field3<br>
- * <b>includeFieldsStr</b>: List of comma separated fields to be replaced/added to the input tuple. Ex: Field1,Field2,Field3<br>
- * <b>store</b>: Specify the type of loader for looking data<br>
- * <br>
- *
- *
- * @displayName Abstract Enrichment Operator
- * @tags Enrichment
- * @param <INPUT> Type of tuples which are received by this operator</T>
- * @param <OUTPUT> Type of tuples which are emitted by this operator</T>
- * @since 2.1.0
- */
-public abstract class AbstractEnrichmentOperator<INPUT, OUTPUT> extends BaseOperator
-{
- /**
- * Keep lookup data cache for fast access.
- */
- private transient CacheManager cacheManager;
-
- private transient CacheStore primaryCache = new CacheStore();
-
- private int entryExpiryDurationInMillis = 24 * 60 * 60 * 1000;
- private int cacheCleanupInMillis = 24 * 60 * 60 * 1000;
- private int cacheSize = 1024;
-
- public transient DefaultOutputPort<OUTPUT> output = new DefaultOutputPort<OUTPUT>();
-
- @InputPortFieldAnnotation(optional = true)
- public transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
- {
- @Override public void process(INPUT tuple)
- {
- processTuple(tuple);
- }
- };
-
- private EnrichmentBackup store;
-
- @NotNull
- protected String lookupFieldsStr;
-
- protected String includeFieldsStr;
-
- protected transient List<String> lookupFields = new ArrayList<String>();
- protected transient List<String> includeFields = new ArrayList<String>();
-
- protected void processTuple(INPUT tuple) {
- Object key = getKey(tuple);
- if(key != null) {
- Object result = cacheManager.get(key);
- OUTPUT out = convert(tuple, result);
- emitTuple(out);
- }
- }
-
- protected abstract Object getKey(INPUT tuple);
-
- protected void emitTuple(OUTPUT tuple) {
- output.emit(tuple);
- }
-
- /* Add data from cached value to input field */
- protected abstract OUTPUT convert(INPUT in, Object cached);
-
- @Override public void setup(Context.OperatorContext context)
- {
- super.setup(context);
-
- cacheManager = new NullValuesCacheManager();
-
- // set expiration to one day.
- primaryCache.setEntryExpiryDurationInMillis(entryExpiryDurationInMillis);
- primaryCache.setCacheCleanupInMillis(cacheCleanupInMillis);
- primaryCache.setEntryExpiryStrategy(ExpiryType.EXPIRE_AFTER_WRITE);
- primaryCache.setMaxCacheSize(cacheSize);
-
- lookupFields = Arrays.asList(lookupFieldsStr.split(","));
- if (includeFieldsStr != null) {
- includeFields = Arrays.asList(includeFieldsStr.split(","));
- }
-
- try {
- store.setFields(lookupFields, includeFields);
-
- cacheManager.setPrimary(primaryCache);
- cacheManager.setBackup(store);
- cacheManager.initialize();
- } catch (IOException e) {
- throw new RuntimeException("Unable to initialize primary cache", e);
- }
- }
-
- /**
- * Set the type of backup store for storing and searching data.
- */
- public void setStore(EnrichmentBackup store) {
- this.store = store;
- }
-
- public EnrichmentBackup getStore() {
- return store;
- }
-
- public CacheStore getPrimaryCache()
- {
- return primaryCache;
- }
-
- public String getLookupFieldsStr()
- {
- return lookupFieldsStr;
- }
-
- /**
- * Set the lookup fields for quick searching. It would be in comma separated list
- */
- public void setLookupFieldsStr(String lookupFieldsStr)
- {
- this.lookupFieldsStr = lookupFieldsStr;
- }
-
- public String getIncludeFieldsStr()
- {
- return includeFieldsStr;
- }
-
- /**
- * Set the list of comma separated fields to be added/replaced to the incoming tuple.
- */
- public void setIncludeFieldsStr(String includeFieldsStr)
- {
- this.includeFieldsStr = includeFieldsStr;
- }
-
- public void setEntryExpiryDurationInMillis(int entryExpiryDurationInMillis)
- {
- this.entryExpiryDurationInMillis = entryExpiryDurationInMillis;
- }
-
- public void setCacheCleanupInMillis(int cacheCleanupInMillis)
- {
- this.cacheCleanupInMillis = cacheCleanupInMillis;
- }
-
- public void setCacheSize(int cacheSize)
- {
- this.cacheSize = cacheSize;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
deleted file mode 100644
index 9155b13..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/EnrichmentBackup.java
+++ /dev/null
@@ -1,18 +0,0 @@
-/**
- * Copyright (c) 2015 DataTorrent, Inc.
- * All rights reserved.
- */
-
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.db.cache.CacheManager;
-import java.util.List;
-/**
- * @since 3.1.0
- */
-
-public interface EnrichmentBackup extends CacheManager.Backup
-{
- public void setFields(List<String> lookupFields,List<String> includeFields);
- public boolean needRefresh();
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
deleted file mode 100644
index 2effed0..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/FSLoader.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/**
- * Copyright (c) 2015 DataTorrent, Inc.
- * All rights reserved.
- */
-package com.datatorrent.contrib.enrichment;
-
-import com.esotericsoftware.kryo.NotNull;
-import com.google.common.collect.Maps;
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.codehaus.jackson.JsonProcessingException;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.map.ObjectReader;
-import org.codehaus.jackson.type.TypeReference;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-/**
- * @since 3.1.0
- */
-
-public class FSLoader extends ReadOnlyBackup
-{
- @NotNull
- private String fileName;
-
- private transient Path filePath;
- private transient FileSystem fs;
- private transient boolean connected;
-
- private transient static final ObjectMapper mapper = new ObjectMapper();
- private transient static final ObjectReader reader = mapper.reader(new TypeReference<Map<String, Object>>()
- {
- });
- private transient static final Logger logger = LoggerFactory.getLogger(FSLoader.class);
-
- public String getFileName()
- {
- return fileName;
- }
-
- public void setFileName(String fileName)
- {
- this.fileName = fileName;
- }
-
- @Override public Map<Object, Object> loadInitialData()
- {
- Map<Object, Object> result = null;
- FSDataInputStream in = null;
- BufferedReader bin = null;
- try {
- result = Maps.newHashMap();
- in = fs.open(filePath);
- bin = new BufferedReader(new InputStreamReader(in));
- String line;
- while ((line = bin.readLine()) != null) {
- try {
- Map<String, Object> tuple = reader.readValue(line);
- if(CollectionUtils.isEmpty(includeFields)) {
- if(includeFields == null)
- includeFields = new ArrayList<String>();
- for (Map.Entry<String, Object> e : tuple.entrySet()) {
- includeFields.add(e.getKey());
- }
- }
- ArrayList<Object> includeTuple = new ArrayList<Object>();
- for(String s: includeFields) {
- includeTuple.add(tuple.get(s));
- }
- result.put(getKey(tuple), includeTuple);
- } catch (JsonProcessingException parseExp) {
- logger.info("Unable to parse line {}", line);
- }
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- } finally {
- if(bin != null)
- IOUtils.closeQuietly(bin);
- if(in != null)
- IOUtils.closeQuietly(in);
- try {
- fs.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- logger.debug("loading initial data {}", result.size());
- return result;
- }
-
- private Object getKey(Map<String, Object> tuple)
- {
- ArrayList<Object> lst = new ArrayList<Object>();
- for(String key : lookupFields) {
- lst.add(tuple.get(key));
- }
- return lst;
- }
-
- @Override public Object get(Object key)
- {
- return null;
- }
-
- @Override public List<Object> getAll(List<Object> keys)
- {
- return null;
- }
-
- @Override public void connect() throws IOException
- {
- Configuration conf = new Configuration();
- this.filePath = new Path(fileName);
- this.fs = FileSystem.newInstance(filePath.toUri(), conf);
- if (!fs.isFile(filePath))
- throw new IOException("Provided path " + fileName + " is not a file");
- connected = true;
- }
-
- @Override public void disconnect() throws IOException
- {
- if (fs != null)
- fs.close();
- }
-
- @Override public boolean isConnected()
- {
- return connected;
- }
-
- @Override
- public boolean needRefresh() {
- return false;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
deleted file mode 100755
index 7040a7a..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/HBaseLoader.java
+++ /dev/null
@@ -1,128 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.contrib.hbase.HBaseStore;
-import com.google.common.collect.Lists;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.util.Bytes;
-
-/**
- * <p>HBaseLoader extends from {@link HBaseStore} uses HBase to connect and implements EnrichmentBackup interface.</p> <br/>
- *
- * Properties:<br>
- * <b>includeFamilys</b>: List of comma separated families and each family corresponds to the group name of column fields in includeFieldsStr. Ex: Family1,Family2<br>
- * <br>
- *
- * @displayName HBaseLoader
- * @tags Loader
- * @since 2.1.0
- */
-public class HBaseLoader extends HBaseStore implements EnrichmentBackup
-{
- protected transient List<String> includeFields;
- protected transient List<String> lookupFields;
- protected transient List<String> includeFamilys;
-
- protected Object getQueryResult(Object key)
- {
- try {
- Get get = new Get(getRowBytes(((ArrayList)key).get(0)));
- int idx = 0;
- for(String f : includeFields) {
- get.addColumn(Bytes.toBytes(includeFamilys.get(idx++)), Bytes.toBytes(f));
- }
- return getTable().get(get);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- protected ArrayList<Object> getDataFrmResult(Object result)
- {
- Result res = (Result)result;
- if (res == null || res.isEmpty())
- return null;
- ArrayList<Object> columnInfo = new ArrayList<Object>();
-
- if(CollectionUtils.isEmpty(includeFields)) {
- if(includeFields == null) {
- includeFields = new ArrayList<String>();
- includeFamilys.clear();
- includeFamilys = new ArrayList<String>();
- }
- for (KeyValue kv: res.raw()) {
- includeFields.add(new String(kv.getQualifier()));
- includeFamilys.add(new String(kv.getFamily()));
- }
- }
- for(KeyValue kv : res.raw()) {
- columnInfo.add(kv.getValue());
- }
- return columnInfo;
- }
-
- private byte[] getRowBytes(Object key)
- {
- return ((String)key).getBytes();
- }
-
- @Override public void setFields(List<String> lookupFields,List<String> includeFields)
- {
- this.includeFields = includeFields;
- this.lookupFields = lookupFields;
- }
-
- /**
- * Set the familyStr and would be in the form of comma separated list.
- */
- public void setIncludeFamilyStr(String familyStr)
- {
- this.includeFamilys = Arrays.asList(familyStr.split(","));
- }
-
- @Override public boolean needRefresh()
- {
- return false;
- }
-
- @Override public Map<Object, Object> loadInitialData()
- {
- return null;
- }
-
- @Override public Object get(Object key)
- {
- return getDataFrmResult(getQueryResult(key));
- }
-
- @Override public List<Object> getAll(List<Object> keys)
- {
- List<Object> values = Lists.newArrayList();
- for (Object key : keys) {
- values.add(get(key));
- }
- return values;
- }
-
- @Override public void put(Object key, Object value)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void putAll(Map<Object, Object> m)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void remove(Object key)
- {
- throw new RuntimeException("Not supported operation");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
deleted file mode 100644
index 3b1a8cf..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/JDBCLoader.java
+++ /dev/null
@@ -1,158 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.db.jdbc.JdbcStore;
-import com.google.common.collect.Lists;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.commons.collections.CollectionUtils;
-
-/**
- * <p>HBaseLoader extends from {@link JdbcStore} uses JDBC to connect and implements EnrichmentBackup interface.</p> <br/>
- *
- * Properties:<br>
- * <b>queryStmt</b>: Sql Prepared Statement which needs to be executed<br>
- * <b>tableName</b>: JDBC table name<br>
- * <br>
- *
- * @displayName JDBCLoader
- * @tags Loader
- * @since 2.1.0
- */
-public class JDBCLoader extends JdbcStore implements EnrichmentBackup
-{
- protected String queryStmt;
-
- protected String tableName;
-
- protected transient List<String> includeFields;
- protected transient List<String> lookupFields;
-
- protected Object getQueryResult(Object key)
- {
- try {
- PreparedStatement getStatement = getConnection().prepareStatement(queryStmt);
- ArrayList<Object> keys = (ArrayList<Object>) key;
- for (int i = 0; i < keys.size(); i++) {
- getStatement.setObject(i+1, keys.get(i));
- }
- return getStatement.executeQuery();
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
-
- protected ArrayList<Object> getDataFrmResult(Object result) throws RuntimeException
- {
- try {
- ResultSet resultSet = (ResultSet) result;
- if (resultSet.next()) {
- ResultSetMetaData rsdata = resultSet.getMetaData();
- // If the includefields is empty, populate it from ResultSetMetaData
- if(CollectionUtils.isEmpty(includeFields)) {
- if(includeFields == null)
- includeFields = new ArrayList<String>();
- for (int i = 1; i <= rsdata.getColumnCount(); i++) {
- includeFields.add(rsdata.getColumnName(i));
- }
- }
- ArrayList<Object> res = new ArrayList<Object>();
- for(String f : includeFields) {
- res.add(resultSet.getObject(f));
- }
- return res;
- } else
- return null;
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
-
- private String generateQueryStmt()
- {
- String stmt = "select * from " + tableName + " where ";
- for (int i = 0; i < lookupFields.size(); i++) {
- stmt = stmt + lookupFields.get(i) + " = ? ";
- if(i != lookupFields.size() - 1) {
- stmt = stmt + " and ";
- }
- }
- logger.info("generateQueryStmt: {}", stmt);
- return stmt;
- }
-
- public String getQueryStmt()
- {
- return queryStmt;
- }
-
- @Override
- public boolean needRefresh() {
- return false;
- }
-
- /**
- * Set the sql Prepared Statement if the enrichment mechanism is query based.
- */
- public void setQueryStmt(String queryStmt)
- {
- this.queryStmt = queryStmt;
- }
-
- public String getTableName()
- {
- return tableName;
- }
- /**
- * Set the table name.
- */
- public void setTableName(String tableName)
- {
- this.tableName = tableName;
- }
-
- @Override public void setFields(List<String> lookupFields,List<String> includeFields)
- {
- this.includeFields = includeFields;
- this.lookupFields = lookupFields;
- if(queryStmt == null)
- queryStmt = generateQueryStmt();
- }
- @Override public Map<Object, Object> loadInitialData()
- {
- return null;
- }
-
- @Override public Object get(Object key)
- {
- return getDataFrmResult(getQueryResult(key));
- }
-
- @Override public List<Object> getAll(List<Object> keys)
- {
- List<Object> values = Lists.newArrayList();
- for (Object key : keys) {
- values.add(get(key));
- }
- return values;
- }
-
- @Override public void put(Object key, Object value)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void putAll(Map<Object, Object> m)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void remove(Object key)
- {
- throw new RuntimeException("Not supported operation");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
deleted file mode 100644
index 040b5ae..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperator.java
+++ /dev/null
@@ -1,57 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import java.util.ArrayList;
-import java.util.Map;
-
-/**
- *
- * This class takes a HashMap tuple as input and extract the value of the lookupKey configured
- * for this operator. It then does a lookup in file/DB to find matching entry and all key-value pairs
- * specified in the file/DB or based on include fields are added to original tuple.
- *
- * Example
- * The file contains data in json format, one entry per line. during setup entire file is read and
- * kept in memory for quick lookup.
- * If file contains following lines, and operator is configured with lookup key "productId"
- * { "productId": 1, "productCategory": 3 }
- * { "productId": 4, "productCategory": 10 }
- * { "productId": 3, "productCategory": 1 }
- *
- * And input tuple is
- * { amount=10.0, channelId=4, productId=3 }
- *
- * The tuple is modified as below before operator emits it on output port.
- * { amount=10.0, channelId=4, productId=3, productCategory=1 }
- *
- *
- * @displayName MapEnrichment
- * @category Database
- * @tags enrichment, lookup
- *
- * @since 2.1.0
- */
-public class MapEnrichmentOperator extends AbstractEnrichmentOperator<Map<String, Object>, Map<String, Object>>
-{
- @Override protected Object getKey(Map<String, Object> tuple)
- {
- ArrayList<Object> keyList = new ArrayList<Object>();
- for(String key : lookupFields) {
- keyList.add(tuple.get(key));
- }
- return keyList;
- }
-
- @Override protected Map<String, Object> convert(Map<String, Object> in, Object cached)
- {
- if (cached == null)
- return in;
-
- ArrayList<Object> newAttributes = (ArrayList<Object>)cached;
- if(newAttributes != null) {
- for (int i = 0; i < includeFields.size(); i++) {
- in.put(includeFields.get(i), newAttributes.get(i));
- }
- }
- return in;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
deleted file mode 100644
index f668683..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/NullValuesCacheManager.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * Copyright (c) 2015 DataTorrent, Inc.
- * All rights reserved.
- */
-
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.db.cache.CacheManager;
-
-/**
- * @since 3.1.0
- */
-public class NullValuesCacheManager extends CacheManager
-{
-
- private static final NullObject NULL = new NullObject();
- @Override
- public Object get(Object key)
- {
- Object primaryVal = primary.get(key);
- if (primaryVal != null) {
- if (primaryVal == NULL) {
- return null;
- }
-
- return primaryVal;
- }
-
- Object backupVal = backup.get(key);
- if (backupVal != null) {
- primary.put(key, backupVal);
- } else {
- primary.put(key, NULL);
- }
- return backupVal;
-
- }
-
- private static class NullObject
- {
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
deleted file mode 100644
index e707198..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/POJOEnrichmentOperator.java
+++ /dev/null
@@ -1,185 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.api.Context;
-import com.datatorrent.lib.util.PojoUtils;
-import com.datatorrent.lib.util.PojoUtils.Getter;
-import com.datatorrent.lib.util.PojoUtils.Setter;
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-
-import com.esotericsoftware.kryo.NotNull;
-import org.apache.commons.lang3.ClassUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- *
- * This class takes a POJO as input and extract the value of the lookupKey configured
- * for this operator. It then does a lookup in file/DB to find matching entry and all key-value pairs
- * specified in the file/DB or based on include fieldMap are added to original tuple.
- *
- * Properties:<br>
- * <b>inputClass</b>: Class to be loaded for the incoming data type<br>
- * <b>outputClass</b>: Class to be loaded for the emitted data type<br>
- * <br>
- *
- * Example
- * The file contains data in json format, one entry per line. during setup entire file is read and
- * kept in memory for quick lookup.
- * If file contains following lines, and operator is configured with lookup key "productId"
- * { "productId": 1, "productCategory": 3 }
- * { "productId": 4, "productCategory": 10 }
- * { "productId": 3, "productCategory": 1 }
- *
- * And input tuple is
- * { amount=10.0, channelId=4, productId=3 }
- *
- * The tuple is modified as below before operator emits it on output port.
- * { amount=10.0, channelId=4, productId=3, productCategory=1 }
- *
- * @displayName BeanEnrichment
- * @category Database
- * @tags enrichment, lookup
- *
- * @since 2.1.0
- */
-public class POJOEnrichmentOperator extends AbstractEnrichmentOperator<Object, Object> {
-
- private transient static final Logger logger = LoggerFactory.getLogger(POJOEnrichmentOperator.class);
- protected Class inputClass;
- protected Class outputClass;
- private transient List<Getter> getters = new LinkedList<Getter>();
- private transient List<FieldObjectMap> fieldMap = new LinkedList<FieldObjectMap>();
- private transient List<Setter> updateSetter = new LinkedList<Setter>();
-
- @NotNull
- protected String outputClassStr;
-
-
- @Override
- protected Object getKey(Object tuple) {
- ArrayList<Object> keyList = new ArrayList<Object>();
- for(Getter g : getters) {
- keyList.add(g.get(tuple));
- }
- return keyList;
- }
-
- @Override
- protected Object convert(Object in, Object cached) {
- try {
- Object o = outputClass.newInstance();
-
- // Copy the fields from input to output
- for (FieldObjectMap map : fieldMap) {
- map.set.set(o, map.get.get(in));
- }
-
- if (cached == null)
- return o;
-
- if(updateSetter.size() == 0 && includeFields.size() != 0) {
- populateUpdatesFrmIncludeFields();
- }
- ArrayList<Object> newAttributes = (ArrayList<Object>)cached;
- int idx = 0;
- for(Setter s: updateSetter) {
- s.set(o, newAttributes.get(idx));
- idx++;
- }
- return o;
- } catch (InstantiationException e) {
- throw new RuntimeException(e);
- } catch (IllegalAccessException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void setup(Context.OperatorContext context) {
- super.setup(context);
- populateUpdatesFrmIncludeFields();
- }
-
- private void populateGettersFrmLookup()
- {
- for (String fName : lookupFields) {
- Getter f = PojoUtils.createGetter(inputClass, fName, Object.class);
- getters.add(f);
- }
- }
-
- private void populateGettersFrmInput()
- {
- Field[] fields = inputClass.getFields();
- for (Field f : fields) {
- Class c = ClassUtils.primitiveToWrapper(f.getType());
- FieldObjectMap fieldMap = new FieldObjectMap();
- fieldMap.get = PojoUtils.createGetter(inputClass, f.getName(), c);
- try {
- fieldMap.set = PojoUtils.createSetter(outputClass, f.getName(), c);
- } catch (Throwable e) {
- throw new RuntimeException("Failed to initialize Output Class for field: " + f.getName(), e);
- }
- this.fieldMap.add(fieldMap);
- }
- }
-
- private void populateUpdatesFrmIncludeFields() {
- if (this.outputClass == null) {
- logger.debug("Creating output class instance from string: {}", outputClassStr);
- try {
- this.outputClass = this.getClass().getClassLoader().loadClass(outputClassStr);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- }
-
- for (String fName : includeFields) {
- try {
- Field f = outputClass.getField(fName);
- Class c;
- if(f.getType().isPrimitive()) {
- c = ClassUtils.primitiveToWrapper(f.getType());
- } else {
- c = f.getType();
- }
- try {
- updateSetter.add(PojoUtils.createSetter(outputClass, f.getName(), c));
- } catch (Throwable e) {
- throw new RuntimeException("Failed to initialize Output Class for field: " + f.getName(), e);
- }
- } catch (NoSuchFieldException e) {
- throw new RuntimeException("Cannot find field '" + fName + "' in output class", e);
- }
- }
- }
-
- public String getOutputClassStr()
- {
- return outputClassStr;
- }
-
- public void setOutputClassStr(String outputClassStr)
- {
- this.outputClassStr = outputClassStr;
- }
-
- @Override protected void processTuple(Object tuple)
- {
- if (inputClass == null) {
- inputClass = tuple.getClass();
- populateGettersFrmLookup();
- populateGettersFrmInput();
- }
- super.processTuple(tuple);
- }
-
- private class FieldObjectMap
- {
- public Getter get;
- public Setter set;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
deleted file mode 100644
index 3357704..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/ReadOnlyBackup.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Copyright (c) 2015 DataTorrent, Inc.
- * All rights reserved.
- */
-package com.datatorrent.contrib.enrichment;
-
-import java.util.List;
-import java.util.Map;
-/**
- * @since 3.1.0
- */
-
-public abstract class ReadOnlyBackup implements EnrichmentBackup
-{
- protected transient List<String> includeFields;
- protected transient List<String> lookupFields;
-
- @Override public void put(Object key, Object value)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void putAll(Map<Object, Object> m)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void remove(Object key)
- {
- throw new RuntimeException("Not supported operation");
- }
-
- @Override public void setFields(List<String> lookupFields,List<String> includeFields)
- {
- this.includeFields = includeFields;
- this.lookupFields = lookupFields;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java b/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
deleted file mode 100644
index 7d5b4cd..0000000
--- a/contrib/src/main/java/com/datatorrent/contrib/enrichment/package-info.java
+++ /dev/null
@@ -1 +0,0 @@
-package com.datatorrent.contrib.enrichment;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/EmployeeOrder.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/EmployeeOrder.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/EmployeeOrder.java
new file mode 100644
index 0000000..6015435
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/EmployeeOrder.java
@@ -0,0 +1,114 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+public class EmployeeOrder
+{
+ public int OID;
+ public int ID;
+ public double amount;
+ public String NAME;
+ public int AGE;
+ public String ADDRESS;
+ public double SALARY;
+
+ public int getOID()
+ {
+ return OID;
+ }
+
+ public void setOID(int OID)
+ {
+ this.OID = OID;
+ }
+
+ public int getID()
+ {
+ return ID;
+ }
+
+ public void setID(int ID)
+ {
+ this.ID = ID;
+ }
+
+ public int getAGE()
+ {
+ return AGE;
+ }
+
+ public void setAGE(int AGE)
+ {
+ this.AGE = AGE;
+ }
+
+ public String getNAME()
+ {
+ return NAME;
+ }
+
+ public void setNAME(String NAME)
+ {
+ this.NAME = NAME;
+ }
+
+ public double getAmount()
+ {
+ return amount;
+ }
+
+ public void setAmount(double amount)
+ {
+ this.amount = amount;
+ }
+
+ public String getADDRESS()
+ {
+ return ADDRESS;
+ }
+
+ public void setADDRESS(String ADDRESS)
+ {
+ this.ADDRESS = ADDRESS;
+ }
+
+ public double getSALARY()
+ {
+ return SALARY;
+ }
+
+ public void setSALARY(double SALARY)
+ {
+ this.SALARY = SALARY;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "{" +
+ "OID=" + OID +
+ ", ID=" + ID +
+ ", amount=" + amount +
+ ", NAME='" + NAME + '\'' +
+ ", AGE=" + AGE +
+ ", ADDRESS='" + ADDRESS.trim() + '\'' +
+ ", SALARY=" + SALARY +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/FileEnrichmentTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/FileEnrichmentTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/FileEnrichmentTest.java
new file mode 100644
index 0000000..f24a13c
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/FileEnrichmentTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Map;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.apache.commons.io.FileUtils;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.google.common.collect.Maps;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+
+public class FileEnrichmentTest
+{
+
+ @Rule
+ public final TestUtils.TestInfo testInfo = new TestUtils.TestInfo();
+
+ @Test
+ public void testEnrichmentOperator() throws IOException, InterruptedException
+ {
+ URL origUrl = this.getClass().getResource("/productmapping.txt");
+
+ URL fileUrl = new URL(this.getClass().getResource("/").toString() + "productmapping1.txt");
+ FileUtils.deleteQuietly(new File(fileUrl.getPath()));
+ FileUtils.copyFile(new File(origUrl.getPath()), new File(fileUrl.getPath()));
+
+ MapEnricher oper = new MapEnricher();
+ FSLoader store = new FSLoader();
+ store.setFileName(fileUrl.toString());
+ oper.setLookupFields(Arrays.asList("productId"));
+ oper.setIncludeFields(Arrays.asList("productCategory"));
+ oper.setStore(store);
+
+ oper.setup(null);
+
+ /* File contains 6 entries, but operator one entry is duplicate,
+ * so cache should contains only 5 entries after scanning input file.
+ */
+ //Assert.assertEquals("Number of mappings ", 7, oper.cache.size());
+
+ CollectorTestSink<Map<String, Object>> sink = new CollectorTestSink<>();
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ CollectorTestSink<Object> tmp = (CollectorTestSink)sink;
+ oper.output.setSink(tmp);
+
+ oper.activate(null);
+
+ oper.beginWindow(0);
+ Map<String, Object> tuple = Maps.newHashMap();
+ tuple.put("productId", 3);
+ tuple.put("channelId", 4);
+ tuple.put("amount", 10.0);
+
+ Kryo kryo = new Kryo();
+ oper.input.process(kryo.copy(tuple));
+
+ oper.endWindow();
+
+ oper.deactivate();
+
+ /* Number of tuple, emitted */
+ Assert.assertEquals("Number of tuple emitted ", 1, sink.collectedTuples.size());
+ Map<String, Object> emitted = sink.collectedTuples.iterator().next();
+
+ /* The fields present in original event is kept as it is */
+ Assert.assertEquals("Number of fields in emitted tuple", 4, emitted.size());
+ Assert.assertEquals("value of productId is 3", tuple.get("productId"), emitted.get("productId"));
+ Assert.assertEquals("value of channelId is 4", tuple.get("channelId"), emitted.get("channelId"));
+ Assert.assertEquals("value of amount is 10.0", tuple.get("amount"), emitted.get("amount"));
+
+ /* Check if productCategory is added to the event */
+ Assert.assertEquals("productCategory is part of tuple", true, emitted.containsKey("productCategory"));
+ Assert.assertEquals("value of product category is 1", 5, emitted.get("productCategory"));
+ Assert.assertTrue(emitted.get("productCategory") instanceof Integer);
+ }
+}
+
[4/4] incubator-apex-malhar git commit: Merge branch
'APEXMALHAR-2023_Enrichment' of
https://github.com/chinmaykolhatkar/incubator-apex-malhar
Posted by sa...@apache.org.
Merge branch 'APEXMALHAR-2023_Enrichment' of https://github.com/chinmaykolhatkar/incubator-apex-malhar
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/79c643b9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/79c643b9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/79c643b9
Branch: refs/heads/master
Commit: 79c643b958ab2db64bacd576edf9ea871b0ff5e9
Parents: 3e09d7b 9600edd
Author: Sandeep Deshmukh <sa...@datatorrent.com>
Authored: Tue May 10 14:26:10 2016 +0530
Committer: Sandeep Deshmukh <sa...@datatorrent.com>
Committed: Tue May 10 14:26:10 2016 +0530
----------------------------------------------------------------------
.../contrib/enrich/AbstractEnricher.java | 321 +++++++++++++++++++
.../contrib/enrich/BackendLoader.java | 42 +++
.../datatorrent/contrib/enrich/FSLoader.java | 184 +++++++++++
.../datatorrent/contrib/enrich/JDBCLoader.java | 201 ++++++++++++
.../datatorrent/contrib/enrich/MapEnricher.java | 138 ++++++++
.../contrib/enrich/NullValuesCacheManager.java | 60 ++++
.../contrib/enrich/POJOEnricher.java | 287 +++++++++++++++++
.../contrib/enrich/ReadOnlyBackup.java | 61 ++++
.../contrib/enrich/EmployeeOrder.java | 114 +++++++
.../contrib/enrich/FileEnrichmentTest.java | 103 ++++++
.../contrib/enrich/JDBCLoaderTest.java | 209 ++++++++++++
.../contrib/enrich/MapEnricherTest.java | 251 +++++++++++++++
.../com/datatorrent/contrib/enrich/Order.java | 71 ++++
.../contrib/enrich/POJOEnricherTest.java | 232 ++++++++++++++
contrib/src/test/resources/productmapping.txt | 100 ++++++
.../com/datatorrent/lib/util/FieldInfo.java | 15 +-
16 files changed, 2388 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
[2/4] incubator-apex-malhar git commit: APEXMALHAR-2023 Adding
Enrichment Operator to Malhar
Posted by sa...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/JDBCLoaderTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/JDBCLoaderTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/JDBCLoaderTest.java
new file mode 100644
index 0000000..f450736
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/JDBCLoaderTest.java
@@ -0,0 +1,209 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.netlet.util.DTThrowable;
+
+public class JDBCLoaderTest
+{
+ static final Logger logger = LoggerFactory.getLogger(JDBCLoaderTest.class);
+
+ public static class TestMeta extends TestWatcher
+ {
+ JDBCLoader dbloader;
+ int[] id = {1, 2, 3, 4};
+ String[] name = {"Paul", "Allen", "Teddy", "Mark"};
+ int[] age = {32, 25, 23, 25};
+ String[] address = {"California", "Texas", "Norway", "Rich-Mond"};
+ double[] salary = {20000.00, 15000.00, 20000.00, 65000.00};
+
+ @Override
+ protected void starting(Description description)
+ {
+ try {
+ dbloader = new JDBCLoader();
+ dbloader.setDatabaseDriver("org.hsqldb.jdbcDriver");
+ dbloader.setDatabaseUrl("jdbc:hsqldb:mem:test;sql.syntax_mys=true");
+ dbloader.setTableName("COMPANY");
+
+ dbloader.connect();
+ createTable();
+ insertRecordsInTable();
+ } catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ private void createTable()
+ {
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+
+ String createTable = "CREATE TABLE " + dbloader.getTableName() + " " +
+ "(ID INT PRIMARY KEY, " +
+ "NAME CHAR(50), " +
+ "AGE INT, " +
+ "ADDRESS CHAR(50), " +
+ "SALARY REAL)";
+ logger.debug(createTable);
+ stmt.executeUpdate(createTable);
+
+ logger.debug("Table created successfully...");
+ } catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ private void insertRecordsInTable()
+ {
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+ String tbName = dbloader.getTableName();
+
+ for (int i = 0; i < id.length; i++) {
+ String sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
+ "VALUES (" + id[i] + ", '" + name[i] + "', " + age[i] + ", '" + address[i] + "', " + salary[i] + " );";
+ stmt.executeUpdate(sql);
+ }
+ } catch (Throwable e) {
+ DTThrowable.rethrow(e);
+ }
+
+ }
+
+ private void cleanTable()
+ {
+ String sql = "DROP TABLE " + dbloader.tableName;
+ try {
+ Statement stmt = dbloader.getConnection().createStatement();
+ stmt.executeUpdate(sql);
+ logger.debug("Table deleted successfully...");
+ } catch (SQLException e) {
+ DTThrowable.rethrow(e);
+ }
+ }
+
+ @Override
+ protected void finished(Description description)
+ {
+ cleanTable();
+ }
+ }
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ @Test
+ public void testMysqlDBLookup() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ ArrayList<FieldInfo> lookupKeys = new ArrayList<>();
+ lookupKeys.add(new FieldInfo("ID", "ID", FieldInfo.SupportType.INTEGER));
+ ArrayList<FieldInfo> includeKeys = new ArrayList<>();
+ includeKeys.add(new FieldInfo("NAME", "NAME", FieldInfo.SupportType.STRING));
+ includeKeys.add(new FieldInfo("AGE", "AGE", FieldInfo.SupportType.INTEGER));
+ includeKeys.add(new FieldInfo("ADDRESS", "ADDRESS", FieldInfo.SupportType.STRING));
+
+ testMeta.dbloader.setFieldInfo(lookupKeys, includeKeys);
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<>();
+ keys.add(4);
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>)testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("NAME", "Mark", columnInfo.get(0).toString().trim());
+ Assert.assertEquals("AGE", 25, columnInfo.get(1));
+ Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(2).toString().trim());
+ }
+
+ @Test
+ public void testMysqlDBLookupIncludeAllKeys() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ ArrayList<FieldInfo> lookupKeys = new ArrayList<>();
+ lookupKeys.add(new FieldInfo("ID", "ID", FieldInfo.SupportType.INTEGER));
+
+ ArrayList<FieldInfo> includeKeys = new ArrayList<>();
+ includeKeys.add(new FieldInfo("ID", "ID", FieldInfo.SupportType.INTEGER));
+ includeKeys.add(new FieldInfo("NAME", "NAME", FieldInfo.SupportType.STRING));
+ includeKeys.add(new FieldInfo("AGE", "AGE", FieldInfo.SupportType.INTEGER));
+ includeKeys.add(new FieldInfo("ADDRESS", "ADDRESS", FieldInfo.SupportType.STRING));
+ includeKeys.add(new FieldInfo("SALARY", "SALARY", FieldInfo.SupportType.DOUBLE));
+
+ testMeta.dbloader.setFieldInfo(lookupKeys, includeKeys);
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add(4);
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>)testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("ID", 4, columnInfo.get(0));
+ Assert.assertEquals("NAME", "Mark", columnInfo.get(1).toString().trim());
+ Assert.assertEquals("AGE", 25, columnInfo.get(2));
+ Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(3).toString().trim());
+ Assert.assertEquals("SALARY", 65000.0, columnInfo.get(4));
+ }
+
+ @Test
+ public void testMysqlDBQuery() throws Exception
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ testMeta.dbloader
+ .setQueryStmt("Select id, name from " + testMeta.dbloader.getTableName() + " where AGE = ? and ADDRESS = ?");
+
+ latch.await(1000, TimeUnit.MILLISECONDS);
+
+ ArrayList<FieldInfo> includeKeys = new ArrayList<>();
+ includeKeys.add(new FieldInfo("ID", "ID", FieldInfo.SupportType.INTEGER));
+ includeKeys.add(new FieldInfo("NAME", "NAME", FieldInfo.SupportType.STRING));
+
+ testMeta.dbloader.setFieldInfo(null, includeKeys);
+
+ ArrayList<Object> keys = new ArrayList<Object>();
+ keys.add(25);
+ keys.add("Texas");
+
+ ArrayList<Object> columnInfo = (ArrayList<Object>)testMeta.dbloader.get(keys);
+
+ Assert.assertEquals("ID", 2, columnInfo.get(0));
+ Assert.assertEquals("NAME", "Allen", columnInfo.get(1).toString().trim());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/MapEnricherTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/MapEnricherTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/MapEnricherTest.java
new file mode 100644
index 0000000..7323f0d
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/MapEnricherTest.java
@@ -0,0 +1,251 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.google.common.collect.Maps;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.io.ConsoleOutputOperator;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.TestUtils;
+
+public class MapEnricherTest
+{
+ @Test
+ public void includeAllKeys()
+ {
+ MapEnricher oper = new MapEnricher();
+ oper.setStore(new MemoryStore());
+ oper.setLookupFields(Arrays.asList("In1"));
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ Map<String, Object> inMap = Maps.newHashMap();
+ inMap.put("In1", "Value1");
+ inMap.put("In2", "Value2");
+
+ oper.activate(null);
+ oper.beginWindow(1);
+ oper.input.process(inMap);
+ oper.endWindow();
+ oper.deactivate();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, C=Val_C, In2=Value2, In1=Value3}",
+ sink.collectedTuples.get(0).toString());
+ }
+
+ @Test
+ public void includeSelectedKeys()
+ {
+ MapEnricher oper = new MapEnricher();
+ oper.setStore(new MemoryStore());
+ oper.setLookupFields(Arrays.asList("In1"));
+ oper.setIncludeFields(Arrays.asList("A", "B"));
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ Map<String, Object> inMap = Maps.newHashMap();
+ inMap.put("In1", "Value1");
+ inMap.put("In2", "Value2");
+
+ oper.activate(null);
+ oper.beginWindow(1);
+ oper.input.process(inMap);
+ oper.endWindow();
+ oper.deactivate();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, In2=Value2, In1=Value1}",
+ sink.collectedTuples.get(0).toString());
+ }
+
+ @Test
+ public void testApplication() throws Exception
+ {
+ LocalMode lma = LocalMode.newInstance();
+ Configuration conf = new Configuration(false);
+ lma.prepareDAG(new EnrichApplication(), conf);
+ LocalMode.Controller lc = lma.getController();
+ lc.run(10000);// runs for 10 seconds and quits
+ }
+
+ public static class EnrichApplication implements StreamingApplication
+ {
+ @Override
+ public void populateDAG(DAG dag, Configuration configuration)
+ {
+ RandomMapGenerator input = dag.addOperator("Input", RandomMapGenerator.class);
+ MapEnricher enrich = dag.addOperator("Enrich", MapEnricher.class);
+ ConsoleOutputOperator console = dag.addOperator("Console", ConsoleOutputOperator.class);
+ console.setSilent(true);
+
+ List<String> includeFields = new ArrayList<>();
+ includeFields.add("A");
+ includeFields.add("B");
+ List<String> lookupFields = new ArrayList<>();
+ lookupFields.add("In1");
+
+ enrich.setStore(new MemoryStore());
+ enrich.setIncludeFields(includeFields);
+ enrich.setLookupFields(lookupFields);
+
+ dag.addStream("S1", input.output, enrich.input);
+ dag.addStream("S2", enrich.output, console.input);
+ }
+ }
+
+ public static class RandomMapGenerator extends BaseOperator implements InputOperator
+ {
+ private int key = 0;
+
+ public final transient DefaultOutputPort output = new DefaultOutputPort();
+
+ @Override
+ public void emitTuples()
+ {
+ Map<String, String> map = new HashMap<>();
+ map.put("In" + (key + 1), "Value" + (key + 1));
+ map.put("In2", "Value3");
+ output.emit(map);
+ }
+ }
+
+ private static class MemoryStore implements BackendLoader
+ {
+ static Map<String, Map> returnData = Maps.newHashMap();
+ private List<FieldInfo> includeFieldInfo;
+
+ static {
+ Map<String, String> map = Maps.newHashMap();
+ map.put("A", "Val_A");
+ map.put("B", "Val_B");
+ map.put("C", "Val_C");
+ map.put("In1", "Value3");
+ returnData.put("Value1", map);
+
+ map = Maps.newHashMap();
+ map.put("A", "Val_A_1");
+ map.put("B", "Val_B_1");
+ map.put("C", "Val_C");
+ map.put("In1", "Value3");
+ returnData.put("Value2", map);
+ }
+
+ @Override
+ public Map<Object, Object> loadInitialData()
+ {
+ return null;
+ }
+
+ @Override
+ public Object get(Object key)
+ {
+ List<String> keyList = (List<String>)key;
+ Map<String, String> keyValue = returnData.get(keyList.get(0));
+ ArrayList<Object> lst = new ArrayList<Object>();
+
+ if (CollectionUtils.isEmpty(includeFieldInfo)) {
+ if (includeFieldInfo == null) {
+ includeFieldInfo = new ArrayList<>();
+ }
+ for (Map.Entry<String, String> entry : keyValue.entrySet()) {
+ // TODO: Identify the types..
+ includeFieldInfo.add(new FieldInfo(entry.getKey(), entry.getKey(), FieldInfo.SupportType.OBJECT));
+ }
+ }
+
+ for (FieldInfo fieldInfo : includeFieldInfo) {
+ lst.add(keyValue.get(fieldInfo.getColumnName()));
+ }
+
+ return lst;
+ }
+
+ @Override
+ public List<Object> getAll(List<Object> keys)
+ {
+ return null;
+ }
+
+ @Override
+ public void put(Object key, Object value)
+ {
+
+ }
+
+ @Override
+ public void putAll(Map<Object, Object> m)
+ {
+
+ }
+
+ @Override
+ public void remove(Object key)
+ {
+
+ }
+
+ @Override
+ public void connect() throws IOException
+ {
+
+ }
+
+ @Override
+ public void disconnect() throws IOException
+ {
+
+ }
+
+ @Override
+ public boolean isConnected()
+ {
+ return false;
+ }
+
+ @Override
+ public void setFieldInfo(List<FieldInfo> lookupFieldInfo, List<FieldInfo> includeFieldInfo)
+ {
+ this.includeFieldInfo = includeFieldInfo;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/Order.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/Order.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/Order.java
new file mode 100644
index 0000000..52aa698
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/Order.java
@@ -0,0 +1,71 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+/**
+ * Input Class for POJO Enricher test.
+ */
+public class Order
+{
+ public int OID;
+ public int ID;
+ public double amount;
+
+ public Order()
+ {
+ // for kryo
+ }
+
+ public Order(int oid, int id, double amount)
+ {
+ this.OID = oid;
+ this.ID = id;
+ this.amount = amount;
+ }
+
+ public int getOID()
+ {
+ return OID;
+ }
+
+ public void setOID(int OID)
+ {
+ this.OID = OID;
+ }
+
+ public int getID()
+ {
+ return ID;
+ }
+
+ public void setID(int ID)
+ {
+ this.ID = ID;
+ }
+
+ public double getAmount()
+ {
+ return amount;
+ }
+
+ public void setAmount(double amount)
+ {
+ this.amount = amount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrich/POJOEnricherTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrich/POJOEnricherTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrich/POJOEnricherTest.java
new file mode 100644
index 0000000..7a6bc27
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/enrich/POJOEnricherTest.java
@@ -0,0 +1,232 @@
+/**
+ * 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 com.datatorrent.contrib.enrich;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+
+public class POJOEnricherTest extends JDBCLoaderTest
+{
+ @Test
+ public void includeSelectedKeys()
+ {
+ POJOEnricher oper = new POJOEnricher();
+ oper.setStore(testMeta.dbloader);
+ oper.setLookupFields(Arrays.asList("ID"));
+ oper.setIncludeFields(Arrays.asList("NAME", "AGE", "ADDRESS"));
+ oper.outputClass = EmployeeOrder.class;
+ oper.inputClass = Order.class;
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ oper.activate(null);
+
+ oper.beginWindow(1);
+ Order tuple = new Order(3, 4, 700);
+ oper.input.process(tuple);
+ oper.endWindow();
+
+ oper.deactivate();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Ouput Tuple: ",
+ "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=0.0}",
+ sink.collectedTuples.get(0).toString());
+ }
+
+ @Test
+ public void includeAllKeys()
+ {
+ POJOEnricher oper = new POJOEnricher();
+ oper.setStore(testMeta.dbloader);
+ oper.setLookupFields(Arrays.asList("ID"));
+ oper.setIncludeFields(Arrays.asList("NAME", "AGE", "ADDRESS", "SALARY"));
+ oper.outputClass = EmployeeOrder.class;
+ oper.inputClass = Order.class;
+ oper.setup(null);
+
+ CollectorTestSink sink = new CollectorTestSink();
+ TestUtils.setSink(oper.output, sink);
+
+ oper.activate(null);
+
+ oper.beginWindow(1);
+ Order tuple = new Order(3, 4, 700);
+ oper.input.process(tuple);
+ oper.endWindow();
+
+ oper.deactivate();
+
+ Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
+ Assert.assertEquals("Ouput Tuple: ",
+ "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=65000.0}",
+ sink.collectedTuples.get(0).toString());
+ }
+
+ @Test
+ public void testApplication() throws Exception
+ {
+ EnrichApplication enrichApplication = new EnrichApplication(testMeta);
+ enrichApplication.setLoader(testMeta.dbloader);
+
+ LocalMode lma = LocalMode.newInstance();
+ Configuration conf = new Configuration(false);
+ lma.prepareDAG(enrichApplication, conf);
+ LocalMode.Controller lc = lma.getController();
+ lc.run(10000);// runs for 10 seconds and quits
+ }
+
+ public static class EnrichApplication implements StreamingApplication
+ {
+ private final TestMeta testMeta;
+ BackendLoader loader;
+
+ public EnrichApplication(TestMeta testMeta)
+ {
+ this.testMeta = testMeta;
+ }
+
+ @Override
+ public void populateDAG(DAG dag, Configuration configuration)
+ {
+ RandomPOJOGenerator input = dag.addOperator("Input", RandomPOJOGenerator.class);
+ POJOEnricher enrich = dag.addOperator("Enrich", POJOEnricher.class);
+ EnrichVerifier verify = dag.addOperator("Verify", EnrichVerifier.class);
+ verify.address = testMeta.address;
+ verify.age = testMeta.age;
+ verify.names = testMeta.name;
+ verify.salary = testMeta.salary;
+
+ enrich.setStore(loader);
+ ArrayList<String> lookupFields = new ArrayList<>();
+ lookupFields.add("ID");
+ ArrayList<String> includeFields = new ArrayList<>();
+ includeFields.add("NAME");
+ includeFields.add("AGE");
+ includeFields.add("ADDRESS");
+ includeFields.add("SALARY");
+ enrich.setLookupFields(lookupFields);
+ enrich.setIncludeFields(includeFields);
+
+ dag.getMeta(enrich).getMeta(enrich.input).getAttributes().put(Context.PortContext.TUPLE_CLASS, Order.class);
+ dag.getMeta(enrich).getMeta(enrich.output).getAttributes()
+ .put(Context.PortContext.TUPLE_CLASS, EmployeeOrder.class);
+
+ dag.addStream("S1", input.output, enrich.input);
+ dag.addStream("S2", enrich.output, verify.input);
+ }
+
+ public void setLoader(BackendLoader loader)
+ {
+ this.loader = loader;
+ }
+ }
+
+ public static class RandomPOJOGenerator implements InputOperator
+ {
+ public transient DefaultOutputPort<Object> output = new DefaultOutputPort<>();
+ private int idx = 0;
+ private boolean emit = true;
+
+ @Override
+ public void emitTuples()
+ {
+ if (!emit) {
+ return;
+ }
+ idx += idx++ % 4;
+ Order o = new Order(1, idx + 1, 100.00);
+ output.emit(o);
+ if (idx == 3) {
+ emit = false;
+ }
+ }
+
+ @Override
+ public void beginWindow(long l)
+ {
+ emit = true;
+ }
+
+ @Override
+ public void endWindow()
+ {
+
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ }
+
+ @Override
+ public void teardown()
+ {
+ }
+ }
+
+ public static class EnrichVerifier extends BaseOperator
+ {
+ private static final Logger logger = LoggerFactory.getLogger(EnrichVerifier.class);
+ String[] names;
+ int[] age;
+ String[] address;
+ double[] salary;
+
+ private transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ @Override
+ public void process(Object o)
+ {
+ Assert.assertTrue(o instanceof EmployeeOrder);
+ EmployeeOrder order = (EmployeeOrder)o;
+ int id = order.getID();
+ Assert.assertTrue(id >= 1 && id <= 4);
+ Assert.assertEquals(1, order.getOID());
+ Assert.assertEquals(100.00, order.getAmount(), 0);
+
+ Assert.assertEquals(names[id - 1], order.getNAME());
+ Assert.assertEquals(age[id - 1], order.getAGE());
+ Assert.assertEquals(address[id - 1], order.getADDRESS());
+ Assert.assertEquals(salary[id - 1], order.getSALARY(), 0);
+ }
+ };
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
deleted file mode 100644
index 308aa82..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/BeanEnrichmentOperatorTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.testbench.CollectorTestSink;
-import com.datatorrent.lib.util.TestUtils;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class BeanEnrichmentOperatorTest extends JDBCLoaderTest
-{
- public class Order {
- public int OID;
- public int ID;
- public double amount;
-
- public Order(int oid, int id, double amount) {
- this.OID = oid;
- this.ID = id;
- this.amount = amount;
- }
- public int getOID()
- {
- return OID;
- }
-
- public void setOID(int OID)
- {
- this.OID = OID;
- }
-
- public int getID()
- {
- return ID;
- }
-
- public void setID(int ID)
- {
- this.ID = ID;
- }
-
- public double getAmount()
- {
- return amount;
- }
-
- public void setAmount(double amount)
- {
- this.amount = amount;
- }
- }
-
-
- @Test
- public void includeSelectedKeys()
- {
- POJOEnrichmentOperator oper = new POJOEnrichmentOperator();
- oper.setStore(testMeta.dbloader);
- oper.setLookupFieldsStr("ID");
- oper.setIncludeFieldsStr("NAME,AGE,ADDRESS");
- oper.outputClass = EmployeeOrder.class;
- oper.setup(null);
-
- CollectorTestSink sink = new CollectorTestSink();
- TestUtils.setSink(oper.output, sink);
-
- oper.beginWindow(1);
- Order tuple = new Order(3, 4, 700);
- oper.input.process(tuple);
- oper.endWindow();
-
- Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
- Assert.assertEquals("Ouput Tuple: ", "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=0.0}", sink.collectedTuples.get(0).toString());
- }
- @Test
- public void includeAllKeys()
- {
- POJOEnrichmentOperator oper = new POJOEnrichmentOperator();
- oper.setStore(testMeta.dbloader);
- oper.setLookupFieldsStr("ID");
- oper.outputClass = EmployeeOrder.class;
- oper.setup(null);
-
- CollectorTestSink sink = new CollectorTestSink();
- TestUtils.setSink(oper.output, sink);
-
-
- oper.beginWindow(1);
- Order tuple = new Order(3, 4, 700);
- oper.input.process(tuple);
- oper.endWindow();
-
- Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
- Assert.assertEquals("Ouput Tuple: ", "{OID=3, ID=4, amount=700.0, NAME='Mark', AGE=25, ADDRESS='Rich-Mond', SALARY=65000.0}", sink.collectedTuples.get(0).toString());
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
deleted file mode 100644
index 00c9d82..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/EmployeeOrder.java
+++ /dev/null
@@ -1,95 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-// This class is needed for Bean Enrichment Operator testing
-public class EmployeeOrder {
- public int OID;
- public int ID;
- public double amount;
- public String NAME;
- public int AGE;
- public String ADDRESS;
- public double SALARY;
-
- public int getOID()
- {
- return OID;
- }
-
- public void setOID(int OID)
- {
- this.OID = OID;
- }
-
- public int getID()
- {
- return ID;
- }
-
- public void setID(int ID)
- {
- this.ID = ID;
- }
-
- public int getAGE()
- {
- return AGE;
- }
-
- public void setAGE(int AGE)
- {
- this.AGE = AGE;
- }
-
- public String getNAME()
- {
- return NAME;
- }
-
- public void setNAME(String NAME)
- {
- this.NAME = NAME;
- }
-
- public double getAmount()
- {
- return amount;
- }
-
- public void setAmount(double amount)
- {
- this.amount = amount;
- }
-
- public String getADDRESS()
- {
- return ADDRESS;
- }
-
- public void setADDRESS(String ADDRESS)
- {
- this.ADDRESS = ADDRESS;
- }
-
- public double getSALARY()
- {
- return SALARY;
- }
-
- public void setSALARY(double SALARY)
- {
- this.SALARY = SALARY;
- }
-
- @Override public String toString()
- {
- return "{" +
- "OID=" + OID +
- ", ID=" + ID +
- ", amount=" + amount +
- ", NAME='" + NAME + '\'' +
- ", AGE=" + AGE +
- ", ADDRESS='" + ADDRESS.trim() + '\'' +
- ", SALARY=" + SALARY +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
deleted file mode 100644
index 934d73b..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/FileEnrichmentTest.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.testbench.CollectorTestSink;
-import com.datatorrent.lib.util.TestUtils;
-import com.esotericsoftware.kryo.Kryo;
-import com.google.common.collect.Maps;
-import org.apache.commons.io.FileUtils;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URL;
-import java.util.Map;
-
-
-public class FileEnrichmentTest
-{
-
- @Rule public final TestUtils.TestInfo testInfo = new TestUtils.TestInfo();
-
- @Test public void testEnrichmentOperator() throws IOException, InterruptedException
- {
- URL origUrl = this.getClass().getResource("/productmapping.txt");
-
- URL fileUrl = new URL(this.getClass().getResource("/").toString() + "productmapping1.txt");
- FileUtils.deleteQuietly(new File(fileUrl.getPath()));
- FileUtils.copyFile(new File(origUrl.getPath()), new File(fileUrl.getPath()));
-
- MapEnrichmentOperator oper = new MapEnrichmentOperator();
- FSLoader store = new FSLoader();
- store.setFileName(fileUrl.toString());
- oper.setLookupFieldsStr("productId");
- oper.setStore(store);
-
- oper.setup(null);
-
- /* File contains 6 entries, but operator one entry is duplicate,
- * so cache should contains only 5 entries after scanning input file.
- */
- //Assert.assertEquals("Number of mappings ", 7, oper.cache.size());
-
- CollectorTestSink<Map<String, Object>> sink = new CollectorTestSink<Map<String, Object>>();
- @SuppressWarnings({ "unchecked", "rawtypes" }) CollectorTestSink<Object> tmp = (CollectorTestSink) sink;
- oper.output.setSink(tmp);
-
- oper.beginWindow(0);
- Map<String, Object> tuple = Maps.newHashMap();
- tuple.put("productId", 3);
- tuple.put("channelId", 4);
- tuple.put("amount", 10.0);
-
- Kryo kryo = new Kryo();
- oper.input.process(kryo.copy(tuple));
-
- oper.endWindow();
-
- /* Number of tuple, emitted */
- Assert.assertEquals("Number of tuple emitted ", 1, sink.collectedTuples.size());
- Map<String, Object> emitted = sink.collectedTuples.iterator().next();
-
- /* The fields present in original event is kept as it is */
- Assert.assertEquals("Number of fields in emitted tuple", 4, emitted.size());
- Assert.assertEquals("value of productId is 3", tuple.get("productId"), emitted.get("productId"));
- Assert.assertEquals("value of channelId is 4", tuple.get("channelId"), emitted.get("channelId"));
- Assert.assertEquals("value of amount is 10.0", tuple.get("amount"), emitted.get("amount"));
-
- /* Check if productCategory is added to the event */
- Assert.assertEquals("productCategory is part of tuple", true, emitted.containsKey("productCategory"));
- Assert.assertEquals("value of product category is 1", 5, emitted.get("productCategory"));
-
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
deleted file mode 100644
index 07be982..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/HBaseLoaderTest.java
+++ /dev/null
@@ -1,162 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.netlet.util.DTThrowable;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.MasterNotRunningException;
-import org.apache.hadoop.hbase.ZooKeeperConnectionException;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestWatcher;
-import org.junit.runner.Description;
-import org.slf4j.LoggerFactory;
-
-public class HBaseLoaderTest
-{
- static final org.slf4j.Logger logger = LoggerFactory.getLogger(HBaseLoaderTest.class);
-
- public static class TestMeta extends TestWatcher
- {
-
- HBaseLoader dbloader;
- @Override
- protected void starting(Description description)
- {
- try {
- dbloader = new HBaseLoader();
- Configuration conf = HBaseConfiguration.create();
- conf.addResource(new Path("file:///home/chaitanya/hbase-site.xml"));
-
- dbloader.setConfiguration(conf);
- dbloader.setZookeeperQuorum("localhost");
- dbloader.setZookeeperClientPort(2181);
-
- dbloader.setTableName("EMPLOYEE");
-
- dbloader.connect();
- createTable();
- insertRecordsInTable();
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
- }
-
- private void createTable()
- {
- try {
- String[] familys = { "personal", "professional" };
- HBaseAdmin admin = new HBaseAdmin(dbloader.getConfiguration());
- HTableDescriptor tableDesc = new HTableDescriptor(dbloader.getTableName());
- for (int i = 0; i < familys.length; i++) {
- tableDesc.addFamily(new HColumnDescriptor(familys[i]));
- }
- admin.createTable(tableDesc);
-
- logger.debug("Table created successfully...");
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
- }
-
- @SuppressWarnings("deprecation")
- public void addRecord(String rowKey, String family, String qualifier, String value) throws Exception {
- try {
- HTable table = new HTable(dbloader.getConfiguration(), dbloader.getTableName());
- Put put = new Put(Bytes.toBytes(rowKey));
- put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes
- .toBytes(value));
- table.put(put);
- } catch (IOException e) {
- DTThrowable.rethrow(e);
- }
- }
- private void insertRecordsInTable()
- {
- try {
- addRecord("row1", "personal", "name", "raju");
- addRecord("row1", "personal", "city", "hyderabad");
- addRecord("row1", "professional", "designation", "manager");
- addRecord("row1", "professional", "Salary", "50000");
-
- addRecord("row2", "personal", "name", "ravi");
- addRecord("row2", "personal", "city", "Chennai");
- addRecord("row2", "professional", "designation", "SE");
- addRecord("row2", "professional", "Salary", "30000");
-
- addRecord("row3", "personal", "name", "rajesh");
- addRecord("row3", "personal", "city", "Delhi");
- addRecord("row3", "professional", "designation", "E");
- addRecord("row3", "professional", "Salary", "10000");
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
-
- }
-
- private void cleanTable()
- {
- String sql = "delete from " + dbloader.getTableName();
- try {
- HBaseAdmin admin = new HBaseAdmin(dbloader.getConfiguration());
- admin.disableTable(dbloader.getTableName());
- admin.deleteTable(dbloader.getTableName());
- } catch (MasterNotRunningException e) {
- e.printStackTrace();
- } catch (ZooKeeperConnectionException e) {
- e.printStackTrace();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- @Override
- protected void finished(Description description)
- {
- cleanTable();
- }
- }
-
- @Rule
- public TestMeta testMeta = new TestMeta();
-
- @Test
- public void testHBaseLookup() throws Exception
- {
- CountDownLatch latch = new CountDownLatch(1);
-
- ArrayList<String> includeKeys = new ArrayList<String>();
- includeKeys.add("city");
- includeKeys.add("Salary");
- ArrayList<String> lookupKeys = new ArrayList<String>();
- lookupKeys.add("ID");
- testMeta.dbloader.setFields(lookupKeys, includeKeys);
-
- String includeFamilyStr = "personal, professional";
- testMeta.dbloader.setIncludeFamilyStr(includeFamilyStr);
-
- latch.await(1000, TimeUnit.MILLISECONDS);
-
- ArrayList<Object> keys = new ArrayList<Object>();
- keys.add("row2");
-
- ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
-
- Assert.assertEquals("CITY", "Chennai", columnInfo.get(0).toString().trim());
- Assert.assertEquals("Salary", 30000, columnInfo.get(1));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
deleted file mode 100644
index 72cfb88..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/JDBCLoaderTest.java
+++ /dev/null
@@ -1,179 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.netlet.util.DTThrowable;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestWatcher;
-import org.junit.runner.Description;
-
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.slf4j.LoggerFactory;
-
-public class JDBCLoaderTest
-{
- static final org.slf4j.Logger logger = LoggerFactory.getLogger(JDBCLoaderTest.class);
-
- public static class TestMeta extends TestWatcher
- {
- JDBCLoader dbloader;
- @Override
- protected void starting(Description description)
- {
- try {
- dbloader = new JDBCLoader();
- dbloader.setDatabaseDriver("org.hsqldb.jdbcDriver");
- dbloader.setDatabaseUrl("jdbc:hsqldb:mem:test;sql.syntax_mys=true");
- dbloader.setTableName("COMPANY");
-
- dbloader.connect();
- createTable();
- insertRecordsInTable();
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
- }
-
- private void createTable()
- {
- try {
- Statement stmt = dbloader.getConnection().createStatement();
-
- String createTable = "CREATE TABLE IF NOT EXISTS " + dbloader.getTableName() +
- "(ID INT PRIMARY KEY NOT NULL," +
- " NAME TEXT NOT NULL, " +
- " AGE INT NOT NULL, " +
- " ADDRESS CHAR(50), " +
- " SALARY REAL)";
- stmt.executeUpdate(createTable);
-
- logger.debug("Table created successfully...");
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
- }
-
- private void insertRecordsInTable()
- {
- try {
- Statement stmt = dbloader.getConnection().createStatement();
- String tbName = dbloader.getTableName();
-
- String sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
- "VALUES (1, 'Paul', 32, 'California', 20000.00 );";
- stmt.executeUpdate(sql);
-
- sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
- "VALUES (2, 'Allen', 25, 'Texas', 15000.00 );";
- stmt.executeUpdate(sql);
-
- sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
- "VALUES (3, 'Teddy', 23, 'Norway', 20000.00 );";
- stmt.executeUpdate(sql);
-
- sql = "INSERT INTO " + tbName + " (ID,NAME,AGE,ADDRESS,SALARY) " +
- "VALUES (4, 'Mark', 25, 'Rich-Mond', 65000.00 );";
- stmt.executeUpdate(sql);
- }
- catch (Throwable e) {
- DTThrowable.rethrow(e);
- }
-
- }
-
- private void cleanTable()
- {
- String sql = "delete from " + dbloader.tableName;
- try {
- Statement stmt = dbloader.getConnection().createStatement();
- stmt.executeUpdate(sql);
- logger.debug("Table deleted successfully...");
- } catch (SQLException e) {
- DTThrowable.rethrow(e);
- }
- }
-
- @Override
- protected void finished(Description description)
- {
- cleanTable();
- }
- }
-
- @Rule
- public TestMeta testMeta = new TestMeta();
-
- @Test
- public void testMysqlDBLookup() throws Exception
- {
- CountDownLatch latch = new CountDownLatch(1);
-
- ArrayList<String> lookupKeys = new ArrayList<String>();
- lookupKeys.add("ID");
- ArrayList<String> includeKeys = new ArrayList<String>();
- includeKeys.add("NAME");
- includeKeys.add("AGE");
- includeKeys.add("ADDRESS");
- testMeta.dbloader.setFields(lookupKeys, includeKeys);
-
- latch.await(1000, TimeUnit.MILLISECONDS);
-
- ArrayList<Object> keys = new ArrayList<Object>();
- keys.add("4");
-
- ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
-
- Assert.assertEquals("NAME", "Mark", columnInfo.get(0).toString().trim());
- Assert.assertEquals("AGE", 25, columnInfo.get(1));
- Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(2).toString().trim());
- }
-
- @Test
- public void testMysqlDBLookupIncludeAllKeys() throws Exception
- {
- CountDownLatch latch = new CountDownLatch(1);
-
- ArrayList<String> lookupKeys = new ArrayList<String>();
- lookupKeys.add("ID");
- ArrayList<String> includeKeys = new ArrayList<String>();
- testMeta.dbloader.setFields(lookupKeys, includeKeys);
-
- latch.await(1000, TimeUnit.MILLISECONDS);
-
- ArrayList<Object> keys = new ArrayList<Object>();
- keys.add("4");
-
- ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
-
- Assert.assertEquals("ID", 4, columnInfo.get(0));
- Assert.assertEquals("NAME", "Mark", columnInfo.get(1).toString().trim());
- Assert.assertEquals("AGE", 25, columnInfo.get(2));
- Assert.assertEquals("ADDRESS", "Rich-Mond", columnInfo.get(3).toString().trim());
- Assert.assertEquals("SALARY", 65000.0, columnInfo.get(4));
- }
-
- @Test
- public void testMysqlDBQuery() throws Exception
- {
- CountDownLatch latch = new CountDownLatch(1);
-
- testMeta.dbloader.setQueryStmt("Select id, name from " + testMeta.dbloader.getTableName() + " where AGE = ? and ADDRESS = ?");
-
- latch.await(1000, TimeUnit.MILLISECONDS);
-
- ArrayList<Object> keys = new ArrayList<Object>();
- keys.add("25");
- keys.add("Texas");
-
- ArrayList<Object> columnInfo = (ArrayList<Object>) testMeta.dbloader.get(keys);
-
- Assert.assertEquals("ID", 2, columnInfo.get(0));
- Assert.assertEquals("NAME", "Allen", columnInfo.get(1).toString().trim());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
deleted file mode 100644
index 845fe80..0000000
--- a/contrib/src/test/java/com/datatorrent/contrib/enrichment/MapEnrichmentOperatorTest.java
+++ /dev/null
@@ -1,152 +0,0 @@
-package com.datatorrent.contrib.enrichment;
-
-import com.datatorrent.lib.testbench.CollectorTestSink;
-import com.datatorrent.lib.util.TestUtils;
-import com.google.common.collect.Maps;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.commons.collections.CollectionUtils;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class MapEnrichmentOperatorTest
-{
- @Test
- public void includeAllKeys()
- {
- MapEnrichmentOperator oper = new MapEnrichmentOperator();
- oper.setStore(new MemoryStore());
- oper.setLookupFieldsStr("In1");
- oper.setup(null);
-
- CollectorTestSink sink = new CollectorTestSink();
- TestUtils.setSink(oper.output, sink);
-
- Map<String, Object> inMap = Maps.newHashMap();
- inMap.put("In1", "Value1");
- inMap.put("In2", "Value2");
-
- oper.beginWindow(1);
- oper.input.process(inMap);
- oper.endWindow();
-
- Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
- Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, C=Val_C, In2=Value2, In1=Value3}", sink.collectedTuples.get(0).toString());
- }
-
- @Test
- public void includeSelectedKeys()
- {
- MapEnrichmentOperator oper = new MapEnrichmentOperator();
- oper.setStore(new MemoryStore());
- oper.setLookupFieldsStr("In1");
- oper.setIncludeFieldsStr("A,B");
- oper.setup(null);
-
- CollectorTestSink sink = new CollectorTestSink();
- TestUtils.setSink(oper.output, sink);
-
- Map<String, Object> inMap = Maps.newHashMap();
- inMap.put("In1", "Value1");
- inMap.put("In2", "Value2");
-
- oper.beginWindow(1);
- oper.input.process(inMap);
- oper.endWindow();
-
- Assert.assertEquals("includeSelectedKeys: Number of tuples emitted: ", 1, sink.collectedTuples.size());
- Assert.assertEquals("Enrich Tuple: ", "{A=Val_A, B=Val_B, In2=Value2, In1=Value1}", sink.collectedTuples.get(0).toString());
- }
-
- private static class MemoryStore implements EnrichmentBackup
- {
- static Map<String, Map> returnData = Maps.newHashMap();
- private List<String> includeFields;
- static {
- Map<String, String> map = Maps.newHashMap();
- map.put("A", "Val_A");
- map.put("B", "Val_B");
- map.put("C", "Val_C");
- map.put("In1", "Value3");
- returnData.put("Value1", map);
-
- map = Maps.newHashMap();
- map.put("A", "Val_A_1");
- map.put("B", "Val_B_1");
- map.put("C", "Val_C");
- map.put("In1", "Value3");
- returnData.put("Value2", map);
- }
-
- @Override public Map<Object, Object> loadInitialData()
- {
- return null;
- }
-
- @Override public Object get(Object key)
- {
- List<String> keyList = (List<String>)key;
- Map<String, String> keyValue = returnData.get(keyList.get(0));
- ArrayList<Object> lst = new ArrayList<Object>();
- if(CollectionUtils.isEmpty(includeFields)) {
- if(includeFields == null)
- includeFields = new ArrayList<String>();
- for (Map.Entry<String, String> e : keyValue.entrySet()) {
- includeFields.add(e.getKey());
- }
- }
- for(String field : includeFields) {
- lst.add(keyValue.get(field));
- }
- return lst;
- }
-
- @Override public List<Object> getAll(List<Object> keys)
- {
- return null;
- }
-
- @Override public void put(Object key, Object value)
- {
-
- }
-
- @Override public void putAll(Map<Object, Object> m)
- {
-
- }
-
- @Override public void remove(Object key)
- {
-
- }
-
- @Override public void connect() throws IOException
- {
-
- }
-
- @Override public void disconnect() throws IOException
- {
-
- }
-
- @Override public boolean isConnected()
- {
- return false;
- }
-
- @Override public void setFields(List<String> lookupFields, List<String> includeFields)
- {
- this.includeFields = includeFields;
-
- }
-
- @Override
- public boolean needRefresh() {
- return false;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/contrib/src/test/resources/productmapping.txt
----------------------------------------------------------------------
diff --git a/contrib/src/test/resources/productmapping.txt b/contrib/src/test/resources/productmapping.txt
new file mode 100755
index 0000000..ece99ff
--- /dev/null
+++ b/contrib/src/test/resources/productmapping.txt
@@ -0,0 +1,100 @@
+{"productCategory": 5, "productId": 0}
+{"productCategory": 4, "productId": 1}
+{"productCategory": 5, "productId": 2}
+{"productCategory": 5, "productId": 3}
+{"productCategory": 5, "productId": 4}
+{"productCategory": 1, "productId": 5}
+{"productCategory": 2, "productId": 6}
+{"productCategory": 4, "productId": 7}
+{"productCategory": 2, "productId": 8}
+{"productCategory": 3, "productId": 9}
+{"productCategory": 1, "productId": 10}
+{"productCategory": 5, "productId": 11}
+{"productCategory": 5, "productId": 12}
+{"productCategory": 1, "productId": 13}
+{"productCategory": 1, "productId": 14}
+{"productCategory": 2, "productId": 15}
+{"productCategory": 3, "productId": 16}
+{"productCategory": 5, "productId": 17}
+{"productCategory": 2, "productId": 18}
+{"productCategory": 2, "productId": 19}
+{"productCategory": 2, "productId": 20}
+{"productCategory": 3, "productId": 21}
+{"productCategory": 2, "productId": 22}
+{"productCategory": 5, "productId": 23}
+{"productCategory": 4, "productId": 24}
+{"productCategory": 1, "productId": 25}
+{"productCategory": 3, "productId": 26}
+{"productCategory": 3, "productId": 27}
+{"productCategory": 3, "productId": 28}
+{"productCategory": 5, "productId": 29}
+{"productCategory": 2, "productId": 30}
+{"productCategory": 3, "productId": 31}
+{"productCategory": 3, "productId": 32}
+{"productCategory": 3, "productId": 33}
+{"productCategory": 1, "productId": 34}
+{"productCategory": 3, "productId": 35}
+{"productCategory": 2, "productId": 36}
+{"productCategory": 1, "productId": 37}
+{"productCategory": 3, "productId": 38}
+{"productCategory": 2, "productId": 39}
+{"productCategory": 1, "productId": 40}
+{"productCategory": 5, "productId": 41}
+{"productCategory": 3, "productId": 42}
+{"productCategory": 5, "productId": 43}
+{"productCategory": 2, "productId": 44}
+{"productCategory": 4, "productId": 45}
+{"productCategory": 5, "productId": 46}
+{"productCategory": 2, "productId": 47}
+{"productCategory": 3, "productId": 48}
+{"productCategory": 5, "productId": 49}
+{"productCategory": 5, "productId": 50}
+{"productCategory": 4, "productId": 51}
+{"productCategory": 5, "productId": 52}
+{"productCategory": 1, "productId": 53}
+{"productCategory": 5, "productId": 54}
+{"productCategory": 4, "productId": 55}
+{"productCategory": 4, "productId": 56}
+{"productCategory": 2, "productId": 57}
+{"productCategory": 4, "productId": 58}
+{"productCategory": 4, "productId": 59}
+{"productCategory": 4, "productId": 60}
+{"productCategory": 1, "productId": 61}
+{"productCategory": 2, "productId": 62}
+{"productCategory": 3, "productId": 63}
+{"productCategory": 5, "productId": 64}
+{"productCategory": 1, "productId": 65}
+{"productCategory": 5, "productId": 66}
+{"productCategory": 5, "productId": 67}
+{"productCategory": 2, "productId": 68}
+{"productCategory": 3, "productId": 69}
+{"productCategory": 3, "productId": 70}
+{"productCategory": 2, "productId": 71}
+{"productCategory": 3, "productId": 72}
+{"productCategory": 4, "productId": 73}
+{"productCategory": 2, "productId": 74}
+{"productCategory": 3, "productId": 75}
+{"productCategory": 3, "productId": 76}
+{"productCategory": 4, "productId": 77}
+{"productCategory": 5, "productId": 78}
+{"productCategory": 4, "productId": 79}
+{"productCategory": 1, "productId": 80}
+{"productCategory": 1, "productId": 81}
+{"productCategory": 1, "productId": 82}
+{"productCategory": 3, "productId": 83}
+{"productCategory": 1, "productId": 84}
+{"productCategory": 5, "productId": 85}
+{"productCategory": 3, "productId": 86}
+{"productCategory": 4, "productId": 87}
+{"productCategory": 1, "productId": 88}
+{"productCategory": 5, "productId": 89}
+{"productCategory": 3, "productId": 90}
+{"productCategory": 5, "productId": 91}
+{"productCategory": 2, "productId": 92}
+{"productCategory": 2, "productId": 93}
+{"productCategory": 3, "productId": 94}
+{"productCategory": 1, "productId": 95}
+{"productCategory": 1, "productId": 96}
+{"productCategory": 5, "productId": 97}
+{"productCategory": 3, "productId": 98}
+{"productCategory": 5, "productId": 99}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/9600eddd/library/src/main/java/com/datatorrent/lib/util/FieldInfo.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/com/datatorrent/lib/util/FieldInfo.java b/library/src/main/java/com/datatorrent/lib/util/FieldInfo.java
index e07bd04..d70ca0c 100644
--- a/library/src/main/java/com/datatorrent/lib/util/FieldInfo.java
+++ b/library/src/main/java/com/datatorrent/lib/util/FieldInfo.java
@@ -20,6 +20,8 @@ package com.datatorrent.lib.util;
import javax.validation.constraints.NotNull;
+import org.apache.commons.lang3.ClassUtils;
+
@SuppressWarnings("rawtypes")
/**
* @since 3.3.0
@@ -114,7 +116,7 @@ public class FieldInfo
}
public static enum SupportType {
- BOOLEAN(Boolean.class), SHORT(Short.class), INTEGER(Integer.class), LONG(Long.class), FLOAT(Float.class), DOUBLE(Double.class), STRING(String.class);
+ BOOLEAN(Boolean.class), SHORT(Short.class), INTEGER(Integer.class), LONG(Long.class), FLOAT(Float.class), DOUBLE(Double.class), STRING(String.class), OBJECT(Object.class);
private Class javaType;
@@ -127,6 +129,17 @@ public class FieldInfo
{
return javaType;
}
+
+ public static SupportType getFromJavaType(Class type)
+ {
+ for (SupportType supportType : SupportType.values()) {
+ if (supportType.getJavaType() == ClassUtils.primitiveToWrapper(type)) {
+ return supportType;
+ }
+ }
+
+ return OBJECT;
+ }
}
}