背景: 挖掘重要的關鍵詞問題：數據量較大(百萬量級),單機版的python雖然處理起來簡單，但是時間較長，內存是否溢出還不好說(看機器性能)解決方案: spark 分布式處理原始數據: 一個txt文件，文件格式每一行都是去停用詞、分詞之后的句子代碼如下：import org.apache.spark.ml.feature.{HashingTF,IDF,Tokenizer}val data = sc.textFile(file_path + "xx.txt").toDF("sentence")val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words") val wordsData = tokenizer.transform(data) val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures") val featurizedData = hashingTF.transform(wordsData) val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features") val idfModel = idf.fit(featurizedData) val rescaledData = idfModel.transform(featurizedData) val d1 = rescaledData.select("words","features") val d2 = d1.select(col("words"),col("features").cast("string"))val Df2 = d2.withColumn("features_1",split(col("features"), "],")).select(col("words"),col("features_1").getItem(0).as("col1"),col("features_1").getItem(1).as("col2"))val code = (sentence: String) => {sentence.replace("[","").replace("]","").replace(")","")} val addCol = udf(code)val DF3 = Df2.withColumn("col3", addCol(Df2("col2")))val DF4 = DF3.select(col("words").cast("string"),col("col3"))val DF5 = DF4.withColumn("words_",addCol(DF4("words")))val DF6 = DF5.select("words_","col3")DF6.createOrReplaceTempView("aa")val DF7 = spark.sql("select words_,col3,single_id,single_tim from aa lateral view posexplode(split(words_,',')) t as single_id_index,single_id lateral view posexplode(split(col3,',')) t as single_tim_index,single_tim where single_id_index = single_tim_index")val DF8 = DF7.select(col("single_id"),col("single_tim").cast("Double"))val DF9 = DF8.groupBy("single_id").sum("single_tim") val DF10 = DF9.sort(DF9("sum(single_tim)").desc)DF10.show(5000)巧妙的應用點: 1、應用udf函數 2、在DF6變量的時候，存在一個兩列字符串，需要將字符串中的詞與另一個字符串中的相同位置的值進行對應，并且每行的句子個數還不一樣，如果將spark Dataframe 格式數據轉化成.rdd.map{...}后續操作比較麻煩，用到hive的窗口函數posexplode是一個多行轉多列的方法，而且還可以找到index，大大提高了效率缺點: 代碼沒有整理，看著有點亂

?

總結

以上是生活随笔為你收集整理的spark计算TF-IDF值(中文)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。