如何在火花流中删除其他结果？

Question

我想让字数统计流式传输，只显示我想在 Twitter 上看到的字。

所以，我做了如下的绳子

import java.util.Properties
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
import org.apache.spark.streaming._

import org.apache.log4j._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.twitter._
import twitter4j.TwitterFactory
import twitter4j.conf.ConfigurationBuilder
import java.util.Properties
import org.apache.spark.storage.StorageLevel
import twitter4j.auth.OAuthAuthorization

val appName = "TwitterData"

val ssc = new StreamingContext(sc, Seconds(10))
val hashTags = "XRP"
val cb = new ConfigurationBuilder
val prop = new Properties()

cb.setDebugEnabled(true).setOAuthConsumerKey("key number").setOAuthConsumerSecret("key number").setOAuthAccessToken("key number").setOAuthAccessTokenSecret("key number")

val bld = cb.build()
val tf = new TwitterFactory(bld)
val twitter = tf.getInstance()
val filters = Array(hashTags).toSeq
val auth = new OAuthAuthorization(bld)
val twitterStream = TwitterUtils.createStream(ssc, Some(auth), filters, StorageLevel.MEMORY_ONLY)

twitterStream.cache()

val lines = twitterStream.map(status => status.getText)
lines.print()

val words = lines.flatMap(_.split(" "))
val pairs = words.map(x => {
  if (x == "xrp" || x == "ripple"){
    (x, 1)
  } else {
  }
})

pairs.print()

ssc.start()

它适用于带有 Twitter 的 Spark Streaming，但按照结果，我想删除除我想要获得的结果之外的所有空白。

-------------------------------------------
Time: 1603866040000 ms
-------------------------------------------
@RuleXRP I need 15to25 usd per xrp
RT @Grayscale: 10/27/20 UPDATE: Net Assets Under Management, Holdings per Share, and Market Price per Share for our Investment Products.

T....

-------------------------------------------
Time: 1603866040000 ms
-------------------------------------------
()
()
()
()
()
()
(xrp,1)
()
()
()
...

我该怎么做？如果有什么方法可以得到我想要变得更好的唯一结果然后我的绳索，请告诉我。我需要你的帮助。 我非常感谢您的建议。 谢谢

Answer 1

val pairs = words.map(x => {
  if (x == "xrp" || x == "ripple"){
    (x, 1)
  } else {
  }
})

这会映射您的结果

相反，您可以在映射之前使用过滤器，这会稍微减少您的代码：

val pairs = words
  .filter(x => x == "xrp" || x == "ripple")
  .map(x => (x, 1))