技术问询：如何运行Spark提取Twitter前10热门话题标签的程序

嘿，我看你正在用Spark Streaming做Twitter热门Hashtag的抓取和Top10展示，这里给你补全完整的可运行代码，再梳理几个关键要点帮你顺利跑通：

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.twitter.TwitterUtils
import org.apache.log4j.{Level, Logger}

object PopularHashtags {
  // 配置日志，过滤冗余输出
  def setupLogging() = {
    val rootLogger = Logger.getRootLogger()
    rootLogger.setLevel(Level.ERROR)
    // 降低Spark和Akka的日志级别，避免控制台被无关信息淹没
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("akka").setLevel(Level.WARN)
  }

  def main(args: Array[String]) {
    // 1. 从文本文件加载Twitter API密钥
    // 假设你的配置文件格式为：consumerKey,consumerSecret,accessToken,accessTokenSecret（一行四个值用逗号分隔）
    val credentials = scala.io.Source.fromFile("twitter_credentials.txt").getLines().next().split(",")
    val Array(consumerKey, consumerSecret, accessToken, accessTokenSecret) = credentials

    // 设置系统属性，TwitterUtils会自动读取这些参数
    System.setProperty("twitter4j.oauth.consumerKey", consumerKey.trim)
    System.setProperty("twitter4j.oauth.consumerSecret", consumerSecret.trim)
    System.setProperty("twitter4j.oauth.accessToken", accessToken.trim)
    System.setProperty("twitter4j.oauth.accessTokenSecret", accessTokenSecret.trim)

    // 2. 初始化StreamingContext：本地模式用多线程（local[*]），批处理间隔10秒
    val ssc = new StreamingContext("local[*]", "TwitterPopularHashtags", Seconds(10))
    setupLogging()

    // 3. 创建Twitter实时数据流
    val tweets = TwitterUtils.createStream(ssc, None)

    // 4. 提取所有Hashtag并转换为键值对格式
    val hashtags = tweets.flatMap(status => 
      status.getHashtagEntities.map(ht => "#" + ht.getText.toLowerCase)
    )
    val hashtagCounts = hashtags.map(tag => (tag, 1))

    // 5. 窗口统计：最近60秒内的Hashtag计数，每10秒更新一次结果
    val windowedCounts = hashtagCounts.reduceByKeyAndWindow(
      (a: Int, b: Int) => a + b, 
      Seconds(60), 
      Seconds(10)
    )

    // 6. 排序并提取Top10热门Hashtag
    val top10Hashtags = windowedCounts.transform(rdd => {
      val sortedRDD = rdd.sortBy(_._2, ascending = false)
      sortedRDD.take(10)
    })

    // 7. 打印结果到控制台
    top10Hashtags.print()

    // 启动流处理并等待程序终止
    ssc.start()
    ssc.awaitTermination()
  }
}

• API密钥配置：确保你的twitter_credentials.txt文件没有多余空格，四个密钥的顺序完全匹配代码里的读取逻辑；生产环境建议用环境变量而非明文文件存储密钥。
• 窗口参数调整：代码里用了60秒的统计窗口、10秒的滑动间隔，你可以根据需求修改这两个值（比如改成30秒窗口+5秒滑动），但滑动间隔最好和批处理间隔保持一致。
• 依赖匹配：要确保你的项目依赖和Spark版本对应，比如SBT里的依赖配置：

  libraryDependencies += "org.apache.spark" %% "spark-streaming-twitter" % "2.4.8" // 替换为你的Spark版本
  

• 本地运行注意：本地模式必须用local[*]或者至少2个线程，因为Spark Streaming需要一个线程接收数据，一个线程处理数据。

内容的提问来源于stack exchange，提问作者Anantanuj Dey