You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streams.apache.org by steveblackmon <gi...@git.apache.org> on 2015/03/25 00:30:59 UTC
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
GitHub user steveblackmon opened a pull request:
https://github.com/apache/incubator-streams/pull/200
STREAMS-277 (+STREAMS-278)
updates to make this module use DocumentClassifier and ActivityConverter interfaces and reflection-based DateTime handling and Activity conversion.
tweaks and renames to the tests to match the new model.
This change impacts testing in streams-runtime-pig, so #179 has been merged into this branch.
Note that this will be a breaking change to streams components and pipelines which rely on 'Serializer'-style conversion. The good news is resolving should be as simple as:
ActivityConverterProcessor converter = new ActivityConverterProcessor();
builder.addStreamsProcessor("converter", converter, 4, "provider");
where upstream provider passes in String or ObjectNode representations of POJOs in com.twitter
TwitterEventClassifier and TwitterEventProcessor are no longer necessary.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/steveblackmon/incubator-streams STREAMS-277
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-streams/pull/200.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #200
----
commit db5b1999d5749b05c62ca8ecd0ab1eb6a7d0080a
Author: sblackmon <sb...@w2odigital.com>
Date: 2015-02-10T21:37:59Z
resolves STREAMS-278
note that pigconvertertest.pig now uses StreamsProcessDatumExec rather than StreamsSerializerExec which has been deleted.
commit a3240c84e1dd4988b6ffdc606a5c991cd5ddc34f
Author: sblackmon <sb...@w2odigital.com>
Date: 2015-02-10T21:41:51Z
added AL headers to test scripts
commit 2cc6118f17aaaaa0ced4ea35f3910f90b05b0889
Author: sblackmon <sb...@w2odigital.com>
Date: 2015-02-10T21:35:34Z
updates to make this module use DocumentClassifier and ActivityConverter interfaces and reflection-based DateTime handling and Activity conversion.
tweaks and renames to the tests to match the new model.
this change impacts testing in streams-runtime-pig, so PR submitted for STREAMS-278 must merge first.
resolves STREAMS-276
commit a57d0311b1f38a994c23ee18f399c469f2ecdc6f
Author: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Date: 2015-03-24T23:25:24Z
Merge branch 'STREAMS-278' into STREAMS-277
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by steveblackmon <gi...@git.apache.org>.
Github user steveblackmon commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/200#discussion_r27162296
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.java ---
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.twitter.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Follow;
+import org.apache.streams.twitter.pojo.*;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Ensures twitter documents can be converted to Activity
+ */
+public class TwitterDocumentClassifier implements DocumentClassifier {
+
+ private static ObjectMapper mapper;
+
+ public List<Class> detectClasses(Object document) {
+
+ Preconditions.checkNotNull(document);
+ Preconditions.checkArgument(document instanceof String || document instanceof ObjectNode);
+
+ mapper = new StreamsJacksonMapper(Lists.newArrayList(StreamsTwitterMapper.TWITTER_FORMAT));
+
+ ObjectNode objectNode;
+ try {
+ if( document instanceof String )
+ objectNode = mapper.readValue((String)document, ObjectNode.class);
+ else
+ objectNode = (ObjectNode) document;
+ } catch (IOException e) {
+ e.printStackTrace();
+ return null;
--- End diff --
i agree. will merge a fix.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by steveblackmon <gi...@git.apache.org>.
Github user steveblackmon commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/200#discussion_r27162621
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java ---
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.pojo.Retweet;
+
+import java.io.Serializable;
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+public class TwitterJsonRetweetActivityConverter implements ActivityConverter<Retweet>, Serializable {
+
+ public static Class requiredClass = Retweet.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonRetweetActivityConverter instance = new TwitterJsonRetweetActivityConverter();
+
+ public static TwitterJsonRetweetActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Retweet fromActivity(Activity deserialized) throws ActivityConversionException {
+ return null;
+ }
+
+ @Override
+ public List<Activity> toActivityList(Retweet retweet) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ updateActivity(retweet, activity);
+
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<Retweet> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
--- End diff --
i agree. will switch to throw NIE.
the easiest way to produce a 'post' activity of the original tweet within a retweet, would be a prior-step processor that does nothing but call Retweet.getRetweetedStatus
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-streams/pull/200
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by jfrazee <gi...@git.apache.org>.
Github user jfrazee commented on the pull request:
https://github.com/apache/incubator-streams/pull/200#issuecomment-86125730
:+1:
Note: formString conversions appear to be lost altogether with ActivityConverter model. The functionality is still there in a sense, but the code to do the same task is more verbose now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by steveblackmon <gi...@git.apache.org>.
Github user steveblackmon commented on the pull request:
https://github.com/apache/incubator-streams/pull/200#issuecomment-86219853
You make a good point about the difficulty of running a multi-type conversion without the Processor. I've broken out a helper class that will make it as easy to convert without the Processor as with a Processor. All tests still pass.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by jfrazee <gi...@git.apache.org>.
Github user jfrazee commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/200#discussion_r27142471
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java ---
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.pojo.Retweet;
+
+import java.io.Serializable;
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+public class TwitterJsonRetweetActivityConverter implements ActivityConverter<Retweet>, Serializable {
+
+ public static Class requiredClass = Retweet.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonRetweetActivityConverter instance = new TwitterJsonRetweetActivityConverter();
+
+ public static TwitterJsonRetweetActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Retweet fromActivity(Activity deserialized) throws ActivityConversionException {
+ return null;
+ }
+
+ @Override
+ public List<Activity> toActivityList(Retweet retweet) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ updateActivity(retweet, activity);
+
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<Retweet> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
--- End diff --
This seems inconsistent with fromActivity(), which returns null instead of NotImplementedException. Also, would there be any sense to allowing a conversion on Tweet, since there are frequently Retweets embedded in them?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: STREAMS-277 (+STREAMS-278)
Posted by jfrazee <gi...@git.apache.org>.
Github user jfrazee commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/200#discussion_r27141988
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.java ---
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.twitter.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Follow;
+import org.apache.streams.twitter.pojo.*;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Ensures twitter documents can be converted to Activity
+ */
+public class TwitterDocumentClassifier implements DocumentClassifier {
+
+ private static ObjectMapper mapper;
+
+ public List<Class> detectClasses(Object document) {
+
+ Preconditions.checkNotNull(document);
+ Preconditions.checkArgument(document instanceof String || document instanceof ObjectNode);
+
+ mapper = new StreamsJacksonMapper(Lists.newArrayList(StreamsTwitterMapper.TWITTER_FORMAT));
+
+ ObjectNode objectNode;
+ try {
+ if( document instanceof String )
+ objectNode = mapper.readValue((String)document, ObjectNode.class);
+ else
+ objectNode = (ObjectNode) document;
+ } catch (IOException e) {
+ e.printStackTrace();
+ return null;
--- End diff --
This doesn't quite seem to be the right way to handle a conversion failure. An error or an empty list makes a bit more sense to me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---