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.

https://apps.twitter.com/

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.

https://mvnrepository.com/artifact/org.twitter4j/twitter4j-core/3.0.6

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