使用 Scala Apache Spark 的 Twitter 热门标签

Question

我正在尝试使用 apache spark 和 scala 获取 twitter 流行标签。我可以打印主题标签，但是当我开始使用 reduce 函数计算主题标签时，出现以下错误

network.ConnectionManager：选择器线程被中断！

我在这里添加代码。请帮我解决这个问题。

import java.io._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import StreamingContext._
import org.apache.spark.SparkContext._
import org.apache.spark.streaming.twitter._

object TwitterPopularTags {

  def main(args: Array[String]) {


    val (master, filters) = (args(0), args.slice(5, args.length))

    // Twitter Authentication credentials
    System.setProperty("twitter4j.oauth.consumerKey", "****")
    System.setProperty("twitter4j.oauth.consumerSecret","****")
    System.setProperty("twitter4j.oauth.accessToken", "****")
    System.setProperty("twitter4j.oauth.accessTokenSecret", "****")


    val ssc = new StreamingContext(master, "TwitterPopularTags", Seconds(10),
      System.getenv("SPARK_HOME"), StreamingContext.jarOfClass(this.getClass))

    val tweets = TwitterUtils.createStream(ssc, None)

    val statuses = tweets.map(status => status.getText())

    val words = statuses.flatMap(status => status.split(" "))
        val hashTags = words.filter(word => word.startsWith("#"))


     val counts = hashTags.map(tag => (tag, 1))
                         .reduceByKeyAndWindow(_ + _, _ - _, Seconds(60 * 5), Seconds(10))

    counts.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

[error] (run-main) java.lang.AssertionError: assertion failed: The 检查点目录尚未设置。请用 StreamingContext.checkpoint() 或 SparkContext.checkpoint() 来设置 检查点目录。 java.lang.AssertionError：断言失败： 检查点目录尚未设置。请用 StreamingContext.checkpoint() 或 SparkContext.checkpoint() 来设置 检查点目录。在 scala.Predef$.assert(Predef.scala:179) 在 org.apache.spark.streaming.dstream.DStream.validate(DStream.scala:181) 在 org.apache.spark.streaming.dstream.DStream$$anonfun$validate$10.apply(DStream.scala:227) 在 org.apache.spark.streaming.dstream.DStream$$anonfun$validate$10.apply(DStream.scala:227) 在 scala.collection.immutable.List.foreach(List.scala:318) 在 org.apache.spark.streaming.dstream.DStream.validate(DStream.scala:227) 在 org.apache.spark.streaming.DStreamGraph$$anonfun$start$3.apply(DStreamGraph.scala:47) 在 org.apache.spark.streaming.DStreamGraph$$anonfun$start$3.apply(DStreamGraph.scala:47) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 在 org.apache.spark.streaming.DStreamGraph.start(DStreamGraph.scala:47) 在 org.apache.spark.streaming.scheduler.JobGenerator.startFirstTime(JobGenerator.scala:114) 在 org.apache.spark.streaming.scheduler.JobGenerator.start(JobGenerator.scala:75) 在 org.apache.spark.streaming.scheduler.JobScheduler.start(JobScheduler.scala:67) 在 org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:410) 在 TwitterPopularTags$.main(TwitterPopularTags.scala:77) 在 TwitterPopularTags.main(TwitterPopularTags.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606) [trace] 堆栈 跟踪抑制：运行最后一次编译：运行完整输出。 2007 年 14 月 11 日 20:07:43 INFO dstream.NetworkReceiver$BlockGenerator：块推送 线程中断 14/11/07 20:07:43 INFO network.ConnectionManager: 选择器线程被中断！ java.lang.RuntimeException：非零 退出代码： 1 在 scala.sys.package$.error(package.scala:27​​) [trace] 堆栈跟踪被抑制：运行最后一次编译：运行完整输出。 [错误]（编译：运行）非零退出代码：1 [错误]总时间：41秒， 已于 2014 年 11 月 7 日晚上 8:07:43 完成

这是我在尝试运行上述代码时遇到的错误。

Answer 1

您正在使用reduceByKeyAndWindow，这将强制您在 Spark 中激活检查点。可以查看如何进行这一行操作here