如何使用Spark/Scala中的频率计数从文本文件创建一个bigram？

Question

我想要一个文本文件并创建一个不以点分隔的所有单词的二元组，删除任何特殊字符。我试图用Spark和Scala来做到这一点。

本文：

你好我的朋友。你今天如何
？再见，我的朋友。

应该产生如下：

招呼我，1个
我的朋友，2
如何，1
今天，1
今天轮空，1
再见我，1

Answer 1

对于RDD中的每一行，首先根据'.'进行分割。然后通过分割' '来标记每个产生的子字符串。一旦被标记化，用replaceAll删除特殊字符并转换为小写字母。每个这些子列表都可以使用sliding转换为包含bigrams的字符串数组的迭代器。

然后，按照要求将bigram数组展平为字符串并将其转换为mkString，然后在groupBy和mapValues之间获得每个数的计数。

最后从RDD中平滑，缩小和收集（二元组，count）元组。

val rdd = sc.parallelize(Array("Hello my Friend. How are", 
           "you today? bye my friend.")) 

rdd.map{ 

    // Split each line into substrings by periods 
    _.split('.').map{ substrings => 

     // Trim substrings and then tokenize on spaces 
     substrings.trim.split(' '). 

     // Remove non-alphanumeric characters, using Shyamendra's 
     // clean replacement technique, and convert to lowercase 
     map{_.replaceAll("""\W""", "").toLowerCase()}. 

     // Find bigrams 
     sliding(2) 
    }. 

    // Flatten, and map the bigrams to concatenated strings 
    flatMap{identity}.map{_.mkString(" ")}. 

    // Group the bigrams and count their frequency 
    groupBy{identity}.mapValues{_.size} 

}. 

// Reduce to get a global count, then collect 
flatMap{identity}.reduceByKey(_+_).collect. 

// Format and print 
foreach{x=> println(x._1 + ", " + x._2)} 

you today, 1 
hello my, 1 
my friend, 2 
how are, 1 
bye my, 1 
today bye, 1  

Answer 2

为了从任何标点符号分开整个单词考虑例如

val words = text.split("\\W+") 

它提供在这种情况下

Array[String] = Array(Hello, my, Friend, How, are, you, today, bye, my, friend) 

配对字转换为元组证明更内联用的概念一个二元组，因此考虑例如

for(Array(a,b,_*) <- words.sliding(2).toArray) 
yield (a.toLowerCase(), b.toLowerCase()) 

其通过ohruunuruus产生

Array((hello,my), (my,friend), (friend,How), (how,are), 
     (are,you), (you,today), (today,bye), (bye,my), (my,friend)) 

答案否则传达的简明的方法。

Answer 3

这应该在星火工作：

def bigramsInString(s: String): Array[((String, String), Int)] = { 

    s.split("""\.""")      // split on . 
    .map(_.split(" ")      // split on space 
      .filter(_.nonEmpty)    // remove empty string 
      .map(_.replaceAll("""\W""", "") // remove special chars 
       .toLowerCase) 
      .filter(_.nonEmpty)     
      .sliding(2)      // take continuous pairs 
      .filter(_.size == 2)    // sliding can return partial 
      .map{ case Array(a, b) => ((a, b), 1) }) 
    .flatMap(x => x)       
} 

val rdd = sc.parallelize(Array("Hello my Friend. How are", 
           "you today? bye my friend.")) 

rdd.map(bigramsInString) 
    .flatMap(x => x)    
    .countByKey     // get result in driver memory as Map 
    .foreach{ case ((x, y), z) => println(s"${x} ${y}, ${z}") } 

// my friend, 2 
// how are, 1 
// today bye, 1 
// bye my, 1 
// you today, 1 
// hello my, 1