question Re: Tweet streams with Storm in Archives of Support Questions (Read Only)

Tweet streams with Storm

junnnninho — Thu, 21 Apr 2016 18:20:33 GMT

Hello,

I am pretty new to Storm and I am getting started by trying to process some tweet streams with it. What would be the basic steps to start it?

I am aware of there is a stream api for it (https://dev.twitter.com/streaming/overview), but how would I integrate it with my Storm elements to start making it work.

Any insights appreciated.

Thanks!

Re: Tweet streams with Storm

pvillard — Thu, 21 Apr 2016 18:27:44 GMT

Hi Wellington,

There are few examples out there to use the Twitter API with Storm. You should have a look to Hortonworks tutorials (like this one http://hortonworks.com/hadoop-tutorial/ingesting-processing-real-time-events-apache-storm/) to have explained and comprehensive examples using the sandbox.

If you only want simple code examples, you will find a lot by searching for "github twitter storm" on the internet. A possible example is: https://github.com/pvillard31/storm-twitter

But there are a lot of possibilities depending on how you want to use the Twitter API. For example, you will find some examples that leverages the Trident API in Storm.

Hope that helps!

Re: Tweet streams with Storm

junnnninho — Sat, 23 Apr 2016 09:45:27 GMT

Thanks Pierre. I am starting with this tutorial on my 3 node cluster ( I am not using sandbox) to get familiar with Storm: http://hortonworks.com/hadoop-tutorial/ingesting-processing-real-time-events-apache-storm

I am following all the steps until the point I have to run: [root@sandbox ~]# storm jar storm-starter-0.0.1-storm-0.9.0.1.jar storm.starter.WordCountTopologyWordCount-c storm.starter.WordCountTopologyWordCount-c nimbus.host=sandbox.hortonworks.com

Here I am getting some errors. Should I run this command from any specific folder.

Here is a partial description of the error:

at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:271) [storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:157) [storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at storm.starter.WordCountTopology.main(WordCountTopology.java:77) [storm-starter-0.0.1-storm-0.9.0.1.jar:?]

Caused by: org.apache.thrift7.transport.TTransportException: java.net.ConnectException: Conexão recusada

at org.apache.thrift7.transport.TSocket.open(TSocket.java:187) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at org.apache.thrift7.transport.TFramedTransport.open(TFramedTransport.java:81) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.security.auth.SimpleTransportPlugin.connect(SimpleTransportPlugin.java:102) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.security.auth.TBackoffConnect.doConnectWithRetry(TBackoffConnect.java:48) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

... 8 more

Caused by: java.net.ConnectException: Conexão recusada

at java.net.PlainSocketImpl.socketConnect(Native Method) ~[?:1.8.0_60]

at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350) ~[?:1.8.0_60]

at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206) ~[?:1.8.0_60]

at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) ~[?:1.8.0_60]

at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) ~[?:1.8.0_60]

at java.net.Socket.connect(Socket.java:589) ~[?:1.8.0_60]

at org.apache.thrift7.transport.TSocket.open(TSocket.java:182) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at org.apache.thrift7.transport.TFramedTransport.open(TFramedTransport.java:81) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.security.auth.SimpleTransportPlugin.connect(SimpleTransportPlugin.java:102) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.security.auth.TBackoffConnect.doConnectWithRetry(TBackoffConnect.java:48) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

... 8 more

Exception in thread "main" java.lang.RuntimeException: Could not find leader nimbus from seed hosts [ip-172-31-34-25.sa-east-1.compute.internal]. Did you specify a valid list of nimbus hosts for config nimbus.seeds

at backtype.storm.utils.NimbusClient.getConfiguredClientAs(NimbusClient.java:90)

at backtype.storm.StormSubmitter.submitTopologyAs(StormSubmitter.java:225)

at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:271)

at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:157)

at storm.starter.WordCountTopology.main(WordCountTopology.java:77)

Thanks-

Wellington

Re: Tweet streams with Storm

pvillard — Sat, 23 Apr 2016 17:52:54 GMT

Hi Wellington,

Sounds like Storm is not in a good shape. Is everything green for Storm in Ambari?

Re: Tweet streams with Storm

junnnninho — Sun, 24 Apr 2016 23:05:06 GMT

Thanks Pierre. Storm was in good shape with all green. I have changed the nimbus.host to host and it worked. 🙂 Now that I was able to run my first topology on Storm I am looking into ways to adapt it to the case of Twitter streams. Thanks for the link to https://github.com/pvillard31/storm-twitter. I will use this on myinitial test. I am start clone your project repo and start working on it from my instance. I guess I have some questions regarding the connection to the Twitter api as well. Would you have documentation explaining step by step these two processes?

Merci!

Re: Tweet streams with Storm

pvillard — Sun, 24 Apr 2016 23:24:00 GMT

Regarding Twitter API, I used Twitter4j (doc and examples here: http://twitter4j.org/en/code-examples.html). You will need to create a Twitter App from your twitter account to create access tokens that will be needed by the API (https://apps.twitter.com/). Regarding the topology itself, you can have a look at the README on github which explains the idea (in short: it only gets tweets containing specific keywords, aggregates the number of tweets for a given keyword in a given time period using ticks, and stores this information in both HDFS and Hive).

Note: if one of the provided answers in this thread is answering your initial question, could you mark it as accepted? It helps other users when there are looking for information 😉 Thanks.

Re: Tweet streams with Storm

junnnninho — Mon, 25 Apr 2016 09:02:02 GMT

Thanks Pierre. You info is very useful. I have cloned the project to my instance and I am about to create the .jars. Which classes do you use to build your storm-twitter-0.0.1-SNAPSHOT.jar ?

Re: Tweet streams with Storm

pvillard — Mon, 25 Apr 2016 14:42:24 GMT

To build the jar, you must use maven and run the command "clean package" that will create an uber jar (with all dependencies included). Then you will be able to run your topology using:

storm jar storm-twitter-0.0.1-SNAPSHOT.jar fr.pvillard.storm.topology.Topology <local|cluster> <consumer key> <consumer key secret> <access token key> <access token secret>

where Topology class is the class where I defined the topology and where there is the main method.

Re: Tweet streams with Storm

junnnninho — Tue, 26 Apr 2016 10:16:02 GMT

Thanks Pierre. Sorry for keeping going back to the same point, but this is the first time I use Maven. I am getting the errors below when I try to clean the package. I am not sure if I have to change something on the pom.xml, or where it is located....

[root@ip-172-31-34-25 bin]# ./mvn clean package https://github.com/pvillard31/storm-twitter

[INFO] Scanning for projects...

[INFO] ------------------------------------------------------------------------

[INFO] BUILD FAILURE

[INFO] ------------------------------------------------------------------------

[INFO] Total time: 0.075 s

[INFO] Finished at: 2016-04-25T23:10:05-04:00

[INFO] Final Memory: 5M/115M

[INFO] ------------------------------------------------------------------------

[ERROR] The goal you specified requires a project to execute but there is no POM in this directory (/opt/apache-maven-3.3.9/bin). Please verify you invoked Maven from the correct directory. -> [Help 1]

[ERROR]

[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.

[ERROR] Re-run Maven using the -X switch to enable full debug logging.

[ERROR]

[ERROR] For more information about the errors and possible solutions, please read the following articles:

[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MissingProjectException

Re: Tweet streams with Storm

pvillard — Tue, 26 Apr 2016 14:05:12 GMT

When running a maven command, the best practice is to run it from within the directory containing the pom.xml file (file containing all the Maven instructions for packaging the product). In this case, you should run the command in the clone of the github project where there is the pom file.

cd /path/to/storm-twitter

/opt/apache-maven-3.3.9/bin/mvn clean package

You may also want to add Maven in your path to be able to directly call mvn command.

Re: Tweet streams with Storm

junnnninho — Wed, 27 Apr 2016 07:25:35 GMT

Hey Pierre. I performed a successful build however I am getting an error saying that the Topology class was not found when I run this:

[root@ip-172-31-34-25 storm-twitter]# storm jar storm-twitter-0.0.1-SNAPSHOT.jar fr.pvillard.storm.topology.Topology host=ec2-52-67-8-253.sa-east-1.compute.amazonaws.com kFI3G29IJ5UOMnbe3qmJpDw5L iZszClk61Lfdu6hTxRAIW1STPX1TtbFXpIKlehxHNUGIpMWYFT 140206682-thGEZ8KIYfYbHY9Rzvzu2CO8ry6UBmSEvUe0zOGZ 8uA5F0T4yhnLv16fgrFP4S6W5ETflmGzLd3dPW1chb46v

Error: Not able to locate nor load fr.pvillard.storm.topology.Topology

Should I run the command above from specific folder?

Thanks-

Wellington

Re: Tweet streams with Storm

pvillard — Thu, 28 Apr 2016 15:56:07 GMT

You should run the command where the jar is and it is probably in the target directory under storm-twitter.

There is no argument host=.... This argument is either "local" or "cluster" depending on how you want to run your topology with storm (in a local way to ease debugging for example or in a distributed way).

A final word : you should not post your Twitter credentials, it is private data and could be used by someone else to act on your behalf. I strongly encourage you to regenerate new credentials from the Twitter apps page.

Re: Tweet streams with Storm

junnnninho — Sat, 30 Apr 2016 10:26:35 GMT

Thanks for the tips Pierre. I was able to run the topoly with no errors (apparently):

460 [main] INFO b.s.u.Utils - Using defaults.yaml from resources

523 [main] INFO b.s.u.Utils - Using storm.yaml from resources

584 [main] INFO b.s.u.Utils - Using defaults.yaml from resources

602 [main] INFO b.s.u.Utils - Using storm.yaml from resources

608 [main] INFO b.s.StormSubmitter - Generated ZooKeeper secret payload for MD5-digest: -8800577250600957523:-7254875077049838623

609 [main] INFO b.s.s.a.AuthUtils - Got AutoCreds []

624 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

653 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

654 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

659 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

663 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

668 [main] INFO b.s.u.StormBoundedExponentialBackoffRetry - The baseSleepTimeMs [2000] the maxSleepTimeMs [60000] the maxRetries [5]

676 [main] INFO b.s.StormSubmitter - Uploading topology jar storm-twitter-0.0.1-SNAPSHOT.jar to assigned location: /hadoop/storm/nimbus/inbox/stormjar-cac9d9fd-128d-43d2-b13d-5effadbbbe75.jar

1702 [main] INFO b.s.StormSubmitter - Successfully uploaded topology jar to assigned location: /hadoop/storm/nimbus/inbox/stormjar-cac9d9fd-128d-43d2-b13d-5effadbbbe75.jar

1702 [main] INFO b.s.StormSubmitter - Submitting topology storm-twitter in distributed mode with conf {"topology.message.timeout.secs":120,"storm.zookeeper.topology.auth.scheme":"digest","storm.zookeeper.topology.auth.payload":"-8800577250600957523:-7254875077049838623"}

1939 [main] INFO b.s.StormSubmitter - Finished submitting topology: storm-twitter

I am checking in the Storm UI and things seems to be OK, however not much processing is going on (nothing seems to bee emitting or transferred). I don't have any output stats from the Spout nor errors; Same thing for all the other Bolts. I posted 2 tweets from my account and created the HIVE table indicated on your docs but no resultas have ever appeared.

How often does your spouts collect new streams from the twitter account? My Keys and Access token are all set to ready and write... What do you think might be possible causes for not getting anything here? Any tips around the best way to troubleshoot this kind of thing?

Thanks!

Re: Tweet streams with Storm

pvillard — Sat, 30 Apr 2016 16:00:45 GMT

Try to run the topology in local mode. It will be easier to see what is happening.

Re: Tweet streams with Storm

junnnninho — Sat, 30 Apr 2016 19:58:43 GMT

Thanks Pierre. For some reason when I try to run on local I get these kind of exceptions:

10887 [Thread-20-TweetHdfsBolt] WARN o.a.h.u.NativeCodeLoader - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

11100 [Thread-20-TweetHdfsBolt] ERROR b.s.util - Async loop died!

java.lang.RuntimeException: Error preparing HdfsBolt: java.net.UnknownHostException: mycluster

at org.apache.storm.hdfs.bolt.AbstractHdfsBolt.prepare(AbstractHdfsBolt.java:109) ~[storm-twitter-0.0.1-SNAPSHOT.jar:?]

at backtype.storm.daemon.executor$fn__7245$fn__7258.invoke(executor.clj:746) ~[storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at backtype.storm.util$async_loop$fn__544.invoke(util.clj:473) [storm-core-0.10.0.2.4.0.0-169.jar:0.10.0.2.4.0.0-169]

at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]

at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]

I was wondering if I should perform this setup (cloning the package and configuring the pom.xml) in order to make the stream collection from Twitter work, or if something that has been included on your package: http://twitter4j.org/en/index.html

Thanks!

Re: Tweet streams with Storm

pvillard — Sat, 30 Apr 2016 20:34:27 GMT

No this is because I didn't export every variable out of the code. If you look at the main class describing the topology (https://github.com/pvillard31/storm-twitter/blob/master/src/main/java/fr/pvillard/storm/topology/Topology.java), you will see that I reference the entry points to write into HDFS (line 38), and into Hive (line 36) as well. You should update this class with your own parameters and rebuild the topology using Maven (or better, export the variables as arguments that you give when running the command).

Re: Tweet streams with Storm

junnnninho — Sun, 01 May 2016 07:32:07 GMT

Thanks again. It worked. I have performed the cahnges and also fixed a permission issue around the /storm/ folder on hdfs. Things seem to be working very well now and I don't see any errors on the Storm UI. However when I go to the HIVE view and try to do a simple SELECT * FROM tweet_counts LIMIT 10; here are the exceptions I get (see below). Have you ever run into this? I have been investigating, but not sure where this is coming from...

ps. ambari is the name of the db where I created the tweet_counts table on my HIVE instance.

{"trace":"org.apache.ambari.view.hive.client.HiveErrorStatusException: H170 Unable to fetch results. java.io.IOException: java.io.FileNotFoundException: Path is not a file: /apps/hive/warehouse/ambari.db/tweet_counts\n\tat org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:75)\n\tat org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:61)\n\tat org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocationsInt(FSNamesystem.java:1828)\n\tat org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocations(FSNamesystem.java:1799)\n\tat org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocations(FSNamesystem.java:1712)\n\tat org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.getBlockLocations(NameNodeRpcServer.java:652)\n\tat .......