You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/03/22 00:20:16 UTC
[38/71] [abbrv] git commit: Activating kafka and hdfs modules
Ensuring providers, processors, and persisters are serializable
Activating kafka and hdfs modules
Ensuring providers, processors, and persisters are serializable
git-svn-id: https://svn.apache.org/repos/asf/incubator/streams/branches/STREAMS-26@1571485 13f79535-47bb-0310-9956-ffa450edef68
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/3d7ff8f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/3d7ff8f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/3d7ff8f8
Branch: refs/heads/master
Commit: 3d7ff8f897ab26cac62a29dff06db118e91fa418
Parents: d256daf
Author: sblackmon <sb...@unknown>
Authored: Mon Feb 24 23:20:47 2014 +0000
Committer: sblackmon <sb...@unknown>
Committed: Mon Feb 24 23:20:47 2014 +0000
----------------------------------------------------------------------
pom.xml | 2 +-
streams-contrib/pom.xml | 6 +-
.../streams/hdfs/WebHdfsPersistWriter.java | 25 ++++++--
.../streams/kafka/KafkaPersistReader.java | 29 +++++----
.../streams/kafka/KafkaPersistWriter.java | 22 +++----
.../streams/rss/provider/RssStreamProvider.java | 41 +++++-------
.../twitter/provider/TwitterEventProcessor.java | 66 ++++++++++++++------
.../twitter/provider/TwitterStreamProvider.java | 4 +-
.../provider/TwitterTimelineProvider.java | 10 ++-
.../TwitterJsonDeleteActivitySerializer.java | 2 +
.../TwitterJsonEventActivitySerializer.java | 3 +-
11 files changed, 131 insertions(+), 79 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 9ba1cae..366ae94 100644
--- a/pom.xml
+++ b/pom.xml
@@ -79,7 +79,7 @@
<clojure.version>1.4.0</clojure.version>
<storm.version>0.9.0.1</storm.version>
<kafka.version>0.8.0</kafka.version>
- <zookeeper.version>3.3.4</zookeeper.version>
+ <zookeeper.version>3.4.5-cdh4.5.0</zookeeper.version>
<netty.version>3.8.0.Final</netty.version>
<json-path.version>0.9.0</json-path.version>
</properties>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/pom.xml b/streams-contrib/pom.xml
index 1e1feca..ad559d8 100644
--- a/streams-contrib/pom.xml
+++ b/streams-contrib/pom.xml
@@ -38,15 +38,15 @@
<modules>
<module>streams-persist-console</module>
<module>streams-persist-elasticsearch</module>
- <!--<module>streams-persist-hdfs</module>-->
- <!--<module>streams-persist-kafka</module>-->
+ <module>streams-persist-hdfs</module>
+ <module>streams-persist-kafka</module>
<!--<module>streams-provider-datasift</module>-->
<!--<module>streams-provider-facebook</module>-->
<!--<module>streams-provider-gnip</module>-->
<!--<module>streams-provider-moreover</module>-->
<module>streams-provider-twitter</module>
<!--<module>streams-provider-sysomos</module>-->
- <!--<module>streams-provider-rss</module>-->
+ <module>streams-provider-rss</module>
<!--<module>streams-proxy-semantria</module>-->
</modules>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
index 86ea72f..de87ba5 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
@@ -16,6 +16,8 @@ import org.apache.streams.core.StreamsPersistWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.streams.hdfs.HdfsConfiguration;
+
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.URI;
@@ -278,14 +280,12 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Runnable
.toString();
}
- @Override
public void start() {
connectToWebHDFS();
}
- @Override
public void stop() {
try {
@@ -300,12 +300,10 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Runnable
}
}
- @Override
public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
this.persistQueue = persistQueue;
}
- @Override
public Queue<StreamsDatum> getPersistQueue() {
return persistQueue;
}
@@ -327,4 +325,23 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Runnable
stop();
}
+
+ @Override
+ public void prepare(Object configurationObject) {
+ connectToWebHDFS();
+ }
+
+ @Override
+ public void cleanUp() {
+ try {
+ flush();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ try {
+ close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
index 5b1bfa5..fd0dfdf 100644
--- a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
+++ b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
@@ -17,6 +17,8 @@ import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.streams.kafka.KafkaConfiguration;
+
import java.io.Serializable;
import java.math.BigInteger;
import java.util.List;
@@ -56,17 +58,10 @@ public class KafkaPersistReader implements StreamsPersistReader, Serializable, R
this.persistQueue = persistQueue;
}
- public KafkaPersistReader(KafkaConfiguration config) {
+ public void setConfig(KafkaConfiguration config) {
this.config = config;
- this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
- }
-
- public KafkaPersistReader(KafkaConfiguration config, Queue<StreamsDatum> persistQueue) {
- this.config = config;
- this.persistQueue = persistQueue;
}
- @Override
public void start() {
Properties props = new Properties();
props.setProperty("serializer.encoding", "UTF8");
@@ -86,7 +81,6 @@ public class KafkaPersistReader implements StreamsPersistReader, Serializable, R
}
- @Override
public void stop() {
consumerConnector.shutdown();
while( !executor.isTerminated()) {
@@ -96,18 +90,21 @@ public class KafkaPersistReader implements StreamsPersistReader, Serializable, R
}
}
- @Override
public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
this.persistQueue = persistQueue;
}
- @Override
public Queue<StreamsDatum> getPersistQueue() {
return this.persistQueue;
}
@Override
public StreamsResultSet readAll() {
+ return readCurrent();
+ }
+
+ @Override
+ public StreamsResultSet readCurrent() {
return null;
}
@@ -138,4 +135,14 @@ public class KafkaPersistReader implements StreamsPersistReader, Serializable, R
// once this class can be told when to shutdown by streams, it will run stop
// stop();
}
+
+ @Override
+ public void prepare(Object configurationObject) {
+ start();
+ }
+
+ @Override
+ public void cleanUp() {
+ stop();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
index 14a26f2..83b61c8 100644
--- a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
+++ b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
@@ -42,17 +42,10 @@ public class KafkaPersistWriter implements StreamsPersistWriter, Serializable, R
this.persistQueue = persistQueue;
}
- public KafkaPersistWriter(KafkaConfiguration config) {
+ public void setConfig(KafkaConfiguration config) {
this.config = config;
- this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
}
- public KafkaPersistWriter(KafkaConfiguration config, Queue<StreamsDatum> persistQueue) {
- this.config = config;
- this.persistQueue = persistQueue;
- }
-
- @Override
public void start() {
Properties props = new Properties();
@@ -68,17 +61,14 @@ public class KafkaPersistWriter implements StreamsPersistWriter, Serializable, R
new Thread(new KafkaPersistWriterTask(this)).start();
}
- @Override
public void stop() {
producer.close();
}
- @Override
public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
this.persistQueue = persistQueue;
}
- @Override
public Queue<StreamsDatum> getPersistQueue() {
return this.persistQueue;
}
@@ -106,4 +96,14 @@ public class KafkaPersistWriter implements StreamsPersistWriter, Serializable, R
// stop();
}
+
+ @Override
+ public void prepare(Object configurationObject) {
+ start();
+ }
+
+ @Override
+ public void cleanUp() {
+ stop();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-rss/src/main/java/org/apache/streams/rss/provider/RssStreamProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-rss/src/main/java/org/apache/streams/rss/provider/RssStreamProvider.java b/streams-contrib/streams-provider-rss/src/main/java/org/apache/streams/rss/provider/RssStreamProvider.java
index fa76f67..9ef0bd6 100644
--- a/streams-contrib/streams-provider-rss/src/main/java/org/apache/streams/rss/provider/RssStreamProvider.java
+++ b/streams-contrib/streams-provider-rss/src/main/java/org/apache/streams/rss/provider/RssStreamProvider.java
@@ -31,7 +31,7 @@ import java.util.concurrent.*;
/**
* Created by sblackmon on 12/10/13.
*/
-public class RssStreamProvider /*extends BaseRichSpout*/ implements StreamsProvider, Serializable {
+public class RssStreamProvider implements StreamsProvider, Serializable {
private final static Logger LOGGER = LoggerFactory.getLogger(RssStreamProvider.class);
@@ -85,7 +85,6 @@ public class RssStreamProvider /*extends BaseRichSpout*/ implements StreamsProvi
this.klass = klass;
}
- @Override
public void start() {
Preconditions.checkNotNull(this.klass);
@@ -105,21 +104,21 @@ public class RssStreamProvider /*extends BaseRichSpout*/ implements StreamsProvi
}
- @Override
public void stop() {
for (int i = 0; i < ((config.getFeeds().size() / 5) + 1); i++) {
inQueue.add(RssEventProcessor.TERMINATE);
}
}
- @Override
public Queue<StreamsDatum> getProviderQueue() {
return this.providerQueue;
}
@Override
public StreamsResultSet readCurrent() {
- return null;
+
+ return (StreamsResultSet) providerQueue;
+
}
@Override
@@ -131,28 +130,16 @@ public class RssStreamProvider /*extends BaseRichSpout*/ implements StreamsProvi
public StreamsResultSet readRange(DateTime start, DateTime end) {
return null;
}
-//
-// @Override
-// public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
-// outputFieldsDeclarer.declare(new Fields("document"));
-// outputFieldsDeclarer.declare(new Fields("type"));
-// }
-//
-// @Override
-// public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
-// collector = spoutOutputCollector;
-// run();
-// }
-//
-// @Override
-// public void nextTuple() {
-// try {
-// collector.emit( new Values(outQueue.take(), klass) );
-// } catch (InterruptedException e) {
-// e.printStackTrace();
-// }
-//
-// }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ start();
+ }
+
+ @Override
+ public void cleanUp() {
+ stop();
+ }
private class RssFeedSetupTask implements Runnable {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
index 934d533..871c5f0 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
@@ -4,7 +4,9 @@ import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
import org.apache.streams.pojo.json.Activity;
import org.apache.streams.twitter.pojo.Delete;
import org.apache.streams.twitter.pojo.Retweet;
@@ -16,6 +18,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
@@ -23,7 +26,7 @@ import java.util.concurrent.BlockingQueue;
/**
* Created by sblackmon on 12/10/13.
*/
-public class TwitterEventProcessor implements Runnable {
+public class TwitterEventProcessor implements StreamsProcessor, Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(TwitterEventProcessor.class);
@@ -58,33 +61,25 @@ public class TwitterEventProcessor implements Runnable {
public void run() {
while(true) {
+ String item;
try {
- String item = inQueue.take();
- Thread.sleep(new Random().nextInt(100));
- if(item==TERMINATE) {
+ item = inQueue.poll();
+ if(item instanceof String && item.equals(TERMINATE)) {
LOGGER.info("Terminating!");
- return;
+ break;
}
- // first check for valid json
- ObjectNode node = (ObjectNode)mapper.readTree(item);
+ ObjectNode objectNode = (ObjectNode) mapper.readTree(item);
- // since data is coming from outside provider, we don't know what type the events are
- Class inClass = TwitterEventClassifier.detectClass(item);
+ StreamsDatum rawDatum = new StreamsDatum(objectNode);
- // if the target is string, just pass-through
- if( java.lang.String.class.equals(outClass))
- outQueue.offer(new StreamsDatum(item));
- else {
- // convert to desired format
- Object out = convert(node, inClass, outClass);
-
- if( out != null && validate(out, outClass))
- outQueue.offer(new StreamsDatum(out));
+ for( StreamsDatum entry : process(rawDatum)) {
+ outQueue.offer(entry);
}
} catch (Exception e) {
e.printStackTrace();
+
}
}
}
@@ -161,4 +156,39 @@ public class TwitterEventProcessor implements Runnable {
return valid;
}
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+
+ // first check for valid json
+ ObjectNode node = (ObjectNode) entry.getDocument();
+
+ String json = node.asText();
+
+ // since data is coming from outside provider, we don't know what type the events are
+ Class inClass = TwitterEventClassifier.detectClass(json);
+
+ // if the target is string, just pass-through
+ if( java.lang.String.class.equals(outClass))
+ return Lists.newArrayList(new StreamsDatum(json));
+ else {
+ // convert to desired format
+ Object out = convert(node, inClass, outClass);
+
+ if( out != null && validate(out, outClass))
+ return Lists.newArrayList(new StreamsDatum(out));
+ }
+
+ return Lists.newArrayList();
+
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+
+ }
+
+ @Override
+ public void cleanUp() {
+
+ }
};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProvider.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProvider.java
index 49b54b2..0520b0f 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProvider.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProvider.java
@@ -56,7 +56,7 @@ public class TwitterStreamProvider implements StreamsProvider, Serializable {
protected StreamingEndpoint endpoint;
protected BasicClient client;
- protected ListeningExecutorService executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
+ protected ListeningExecutorService executor;
private static ExecutorService newFixedThreadPoolWithQueueSize(int nThreads, int queueSize) {
return new ThreadPoolExecutor(nThreads, nThreads,
@@ -116,6 +116,8 @@ public class TwitterStreamProvider implements StreamsProvider, Serializable {
@Override
public void prepare(Object o) {
+ executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
+
Preconditions.checkNotNull(this.klass);
Preconditions.checkNotNull(config.getOauth().getConsumerKey());
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
index ebf9dbc..a91a7ec 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
@@ -29,7 +29,7 @@ import java.util.concurrent.*;
/**
* Created by sblackmon on 12/10/13.
*/
-public class TwitterTimelineProvider implements StreamsProvider, Serializable {
+public class TwitterTimelineProvider implements StreamsProvider, Serializable, Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(TwitterTimelineProvider.class);
@@ -57,7 +57,7 @@ public class TwitterTimelineProvider implements StreamsProvider, Serializable {
// return inQueue;
// }
- protected ListeningExecutorService executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
+ protected ListeningExecutorService executor;
protected DateTime start;
protected DateTime end;
@@ -88,8 +88,14 @@ public class TwitterTimelineProvider implements StreamsProvider, Serializable {
this.klass = klass;
}
+ public Queue<StreamsDatum> getProviderQueue() {
+ return this.providerQueue;
+ }
+
public void run() {
+ executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
+
Preconditions.checkNotNull(providerQueue);
Preconditions.checkNotNull(this.klass);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
index f13bf91..804f3b4 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
@@ -8,6 +8,8 @@ import org.apache.streams.pojo.json.Actor;
import org.apache.streams.twitter.pojo.Delete;
import org.apache.streams.twitter.pojo.Tweet;
+import java.io.Serializable;
+
/**
* Created with IntelliJ IDEA.
* User: mdelaet
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/3d7ff8f8/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonEventActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonEventActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonEventActivitySerializer.java
index 83be402..9d53527 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonEventActivitySerializer.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonEventActivitySerializer.java
@@ -15,6 +15,7 @@ import org.apache.streams.pojo.json.Icon;
import org.apache.streams.pojo.json.Provider;
import java.io.IOException;
+import java.io.Serializable;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
@@ -29,7 +30,7 @@ import java.util.Map;
* Time: 9:24 AM
* To change this template use File | Settings | File Templates.
*/
-public abstract class TwitterJsonEventActivitySerializer implements ActivitySerializer<String> {
+public abstract class TwitterJsonEventActivitySerializer implements ActivitySerializer<String>, Serializable {
public static final String DATE_FORMAT = "EEE MMM dd HH:mm:ss Z yyyy";