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 does not discuss anything in regards to Spark streaming techniques.

Create Your Twitter App

To stream Twitter data, you must create a Twitter app first.

You need 4 key values from the “Key and Access Token” in your Twitter app.

1. Consumer Key (API Key)

2. Consumer Secret (API Secret)

3. Access Token              

4. Access Token Secret

5. Optional - Filter Key - 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 to our pom.xml file in the Scala IDE build environment of our 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 a reference to the “twitter4j-core-3.0.6.jar” file in the Scala IDE build of the Eclipse SDK. We can download the “twitter4j-core-3.0.6.jar” jar file from a 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 this was informative!