Friday, 15 June 2018

Twitter Live Streaming with Spark Streaming (Using Scala)


Introduction
In this article we are trying to demonstrate, How to use Spark streaming with Twitter. It’s a demo only article and not discussing anything about spark streaming technique.

Create your twitter Apps
To stream the twitter data, you must create twitter apps first.

You need 4 key values from “Key and Access Token” of Twitter Apps.

1.    Consumer Key (API Key)
2.     Consumer Secret (API Secret)
3.    Access Token              
4.    Access Token Secret

5.    Filter Key   - is optional, if we need to filter any specified subject in twitter





Scala IDE build of Eclipse SDK change the pom.xml dependency
We need to add the Dependency on Pom.xml in Scala IDE build of Eclipse SDK

<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-twitter_2.10</artifactId>
            <version>1.0.0</version>
</dependency>

Adding twitter4j-core-3.0.6.jar
We need to add the reference of “twitter4j-core-3.0.6.jar” file in Scala IDE build of Eclipse SDK. We can download the “twitter4j-core-3.0.6.jar” jar file from MAVEN repository.






Create Scala Code



Scala Code Details
package com.sqlknowledgebank.spark.sparkstreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import twitter4j.conf.ConfigurationBuilder
import twitter4j.auth.OAuthAuthorization
import twitter4j.Status
import org.apache.spark.streaming.twitter.TwitterUtils



object twitterstreaming {
  def main(args: Array[String]) {
    if (args.length < 4) {
      System.err.println("Usage: TwitterData <ConsumerKey><ConsumerSecret><accessToken><accessTokenSecret>" +
        "[<filters>]")

      System.exit(1)
    }
      
    val appName = "TwitterData"
    val conf = new SparkConf()
    conf.setAppName(appName).setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(5))
      
    val Array(consumerKey, consumerSecret, accessToken, accessTokenSecret) = args.take(4)

    val filters = args.takeRight(args.length - 4)
    val cb = new ConfigurationBuilder
      
    cb.setDebugEnabled(true).setOAuthConsumerKey(consumerKey)
      .setOAuthConsumerSecret(consumerSecret)
      .setOAuthAccessToken(accessToken)
      .setOAuthAccessTokenSecret(accessTokenSecret)
        
    val auth = new OAuthAuthorization(cb.build)
    val tweets = TwitterUtils.createStream(ssc, Some(auth), filters)
  
             
    tweets .saveAsTextFiles("tweets", "json")
    ssc.start()
    ssc.awaitTermination()
  }
}

Passing Command Line Parameters to Run







Hope it will be informative.


Posted By: MR. JOYDEEP DAS

No comments:

Post a Comment