如何使用Spark在文本表中查找关键字？

hongyan 发表于 Dev

hongyan

我是Spark的新手。我在HDFS中有两个表。一个表（表1）是一个标记表，由一些文本组成，可以是一些单词或句子。另一个表（表2）具有一个文本列。表1中每一行可能有多个关键字。我的任务是找出表1中表2中文本列的所有匹配关键字，并输出表2中每一行的关键字列表。

问题是我必须遍历表2和表1中的每一行。如果我为表1产生一个大列表，并为表2使用map函数，我仍然必须使用循环来遍历映射中的列表。功能。即使循环不大（万次），驱动程序也会显示JVM内存限制错误。

myTag是表1的标签列表。

def ourMap(line: String, myTag: List[String]): String = {
  var ret = line
  val length = myTag.length
  for (i <- 0 to length - 1) {
    if (line.contains(myTag(i)))
      ret = ret.replaceAll(myTag(i), "_")
  }
  ret
}

val matched = result.map(b => ourMap(b, tagList))

对完成这项任务有什么建议吗？带有或不带有Spark非常感谢！

示例如下：

表格1

row1|Spark  
row2|RDD

表2

row1| Spark is a fast and general engine. RDD supports two types of operations.  
row2| All transformations in Spark are lazy.  
row3| It is for test. I am a sentence.

预期结果：

row1| Spark,RDD  
row2| Spark

主要编辑：

第一个表实际上可能包含句子，而不仅仅是简单的关键字：

row1| Spark  
row2| RDD  
row3| two words  
row4| I am a sentence

以利亚

在这里，请考虑您提供的数据样本：

val table1: Seq[(String, String)] = Seq(("row1", "Spark"), ("row2", "RDD"), ("row3", "Hashmap"))
val table2: Seq[String] = Seq("row1##Spark is a fast and general engine. RDD supports two types of operations.", "row2##All transformations in Spark are lazy.")

val rdd1: RDD[(String, String)] = sc.parallelize(table1)
val rdd2: RDD[(String, String)] = sc.parallelize(table2).map(_.split("##").toList).map(l => (l.head, l.tail(0))).cache

我们将建立第二个数据表的反向索引，并将其连接到第一个表：

val df1: DataFrame = rdd1.toDF("key", "value")
val df2: DataFrame = rdd2.toDF("key", "text")
val df3: DataFrame = rdd2.flatMap { case (row, text) => text.trim.split( """[^\p{IsAlphabetic}]+""")
  .map(word => (word, row))
}.groupByKey.mapValues(_.toSet.toSeq).toDF("word", "index")

import org.apache.spark.sql.functions.explode

val results: RDD[(String, String)] = df3.join(df1, df1("value") === df3("word")).drop("key").drop("value").withColumn("index", explode($"index")).rdd.map {
  case r: Row => (r.getAs[String]("index"), r.getAs[String]("word"))
}.groupByKey.mapValues(i => i.toList.mkString(","))

results.take(2).foreach(println)
// (row1,Spark,RDD)
// (row2,Spark)

主要编辑：

如评论中所述：问题的规格已更改。关键字不再是简单的关键字，它们可能是句子。在这种情况下，这种方法将行不通，这是另一种问题。一种方法是对最近邻居搜索使用局部敏感哈希（LSH）算法。

该算法的实现在这里可用。

不幸的是，该算法及其实现太长，无法在SO上进行讨论。

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。