Over a million developers have joined DZone.
{{announcement.body}}
{{announcement.title}}

Twitter Live Streaming With Spark Streaming (Using Scala)

DZone's Guide to

Twitter Live Streaming With Spark Streaming (Using Scala)

In this post, we go through a quick step-by-step demonstration of how to use Spark streaming techniques with a Twitter application. Let's get to it!

· Big Data Zone ·
Free Resource

How to Simplify Apache Kafka. Get eBook.

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.

Image title

Image title

Create Scala Code

Image title

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

Image title

Hope this was informative!

Topics:
big data ,streaming data ,application data ,twitter apps

Published at DZone with permission of

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}