hadoop-mapreduce-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From siva kumar <siva165...@gmail.com>
Subject spark streaming
Date Fri, 05 Feb 2016 13:20:35 GMT
Hi ,
         Im trying to pull data from twitter and do some processing out of
it.  I found this code

import org.apache.spark.streaming._import
org.apache.spark.streaming.twitter._import
org.apache.spark.storage.StorageLevelimport scala.io.Sourceimport
scala.collection.mutable.HashMapimport java.io.Fileimport
org.apache.log4j.Loggerimport org.apache.log4j.Levelimport
sys.process.stringSeqToProcess
/** Configures the Oauth Credentials for accessing Twitter */def
configureTwitterCredentials(apiKey: String, apiSecret: String,
accessToken: String, accessTokenSecret: String) {
  val configs = new HashMap[String, String] ++= Seq(
    "apiKey" -> apiKey, "apiSecret" -> apiSecret, "accessToken" ->
accessToken, "accessTokenSecret" -> accessTokenSecret)
  println("Configuring Twitter OAuth")
  configs.foreach{ case(key, value) =>
    if (value.trim.isEmpty) {
      throw new Exception("Error setting authentication - value for "
+ key + " not set")
    }
    val fullKey = "twitter4j.oauth." + key.replace("api", "consumer")
    System.setProperty(fullKey, value.trim)
    println("\tProperty " + fullKey + " set as [" + value.trim + "]")
  }
  println()}
// Configure Twitter credentialsval apiKey =
"xxxxxxxxxxxxxxxxxxxxxxxxx"val apiSecret =
"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"val accessToken =
"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"val
accessTokenSecret =
"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"configureTwitterCredentials(apiKey,
apiSecret, accessToken, accessTokenSecret)
import org.apache.spark.streaming.twitter._val ssc = new
StreamingContext(sc, Seconds(2))val tweets =
TwitterUtils.createStream(ssc, None)val twt =
tweets.window(Seconds(60))
case class Tweet(createdAt:Long, text:String)twt.map(status=>
  Tweet(status.getCreatedAt().getTime()/1000,
status.getText())).foreachRDD(rdd=>
  // Below line works only in spark 1.3.0.
  // For spark 1.1.x and spark 1.2.x,
  // use rdd.registerTempTable("tweets") instead.
  rdd.toDF().registerAsTable("tweets"))
twt.print
ssc.start()

But getting the below error

import org.apache.spark.streaming.twitter._
<console>:19: error: object twitter is not a member of package
org.apache.spark.streaming
       import org.apache.spark.streaming.twitter._

I  using spark 1.3.0


The same worked for spark 0.9.0 . Any help?

Mime
View raw message