Spark流媒体-过滤带有地理定位的流媒体后的推文

Question

我是一个初学者，试图使用spark streaming获得推文，使用Scala和一些过滤器关键字。是否有可能在流媒体之后只过滤那些没有地理定位为Null的推文？我正在尝试保存ElasticSearch中的推文。所以，在将tweet地图保存到ElasticSearch之前，我可以过滤那些带有地理定位信息的地图，然后保存它们吗？我正在使用json4s.jsondsl和tweet中的字段创建JSON。这是示例代码

val stream=twitterutils.createstream（ssc,None,filters）val tweetMap=stream.map（status=>{val tweetMap=

      ("location" -> Option(status.getGeoLocation).map(geo => { s"${geo.getLatitude},${geo.getLongitude}" })) ~
      ("UserLang" -> status.getUser.getLang) ~
      ("UserLocation" -> Option(status.getUser.getLocation)) ~
      ("UserName" -> status.getUser.getName) ~
      ("Text" -> status.getText) ~
      ("TextLength" -> status.getText.length) ~
      //Tokenized the tweet message and then filtered only words starting with #
      ("HashTags" -> status.getText.split(" ").filter(_.startsWith("#")).mkString(" ")) ~
      ("PlaceCountry" -> Option(status.getPlace).map (pl => {s"${pl.getCountry}"}))

// Each batch is saved to Elasticsearch 
tweetMap.foreachRDD { tweets => EsSpark.saveToEs(tweets, "sparksender/tweets")) }

//在这一步之前，有没有办法过滤掉“位置”为空的tweets？

我引用了来自GitHub的代码:https://github.com/luvgupta008/screamingtwitter/blob/master/src/main/scala/com/spark/streaming/twitterprismer.scala

Answer 1

查看RDD上的filter方法。接受谓词函数(a:a)=>Boolean。如果返回值为true，则将该元素添加到列表中。如果为false，则不会将该元素添加到列表中。

tweetMap.filter(
  status => Option(status.getGeoLocation) match {
    case Some(_) => true
    case None => false
  })