使用Spark Streaming API测试Twitter

Question

我是Spark流媒体框架的新手，正在尝试处理推特流。我正在编写测试用例，并了解我可以使用Spark StreamingSuite Base，这将帮助我将输入测试为函数流。但我编写了一个函数，它以DStream[状态]作为输入，处理后将DStream[字符串]作为输出。我在StreamingSuite数据库中使用的api是testOperation。

test("Filter only words Starting with #")  {
  val inputTweet = List(List("this is #firstHash"), List("this is #secondHash"), List("this is #thirdHash"))
  val expected = List(List("#firstHash"), List("#secondHash"), List("#thirdHash"))

  testOperation(inputTweet, TransformTweets.getText _, expected, ordered = false)

这是发送输入的函数。。

 def getText(englishTweets: DStream[Status]): DStream[String] = {
    println(englishTweets.toString)

    val hashTags = englishTweets.flatMap(x => x.getText.split(" ").filter(_.startsWith("#")))

    hashTags
  }

但是由于DStream[状态]和DStream[字符串]，我得到了错误“类型不匹配”。我如何模拟流[状态]。

Answer 1

因此，我通过从推特对象工厂（TwitterObjectFactory）的“创建状态”（createStatus）API获取推特状态来解决了这个问题。没有必要模拟TwitterStatus。即使您设法模拟它，也存在序列化问题。因此，这是最好的解决方案：

val rawJson = Source.fromURL(getClass.getResource("/tweetStatus.json")).getLines.mkString
val tweetStatus = TwitterObjectFactory.createStatus(rawJson)

希望这对某人有帮助！