數據挖掘：基於Spark+HanLP實現影視評論關鍵詞抽取(1)

1. 背景

近日項目要求基於爬取的影視評論信息，抽取影視的關鍵字信息。考慮到影視評論數據量較大，因此采用Spark處理框架。關鍵詞提取的處理主要包含分詞+算法抽取兩部分。目前分詞工具包較為主流的，包括哈工大的LTP以及HanLP，而關鍵詞的抽取算法較多，包括TF-IDF、TextRank、互信息等。本次任務主要基於LTP、HanLP、Ac雙數組進行分詞，采用TextRank、互信息以及TF-IDF結合的方式進行關鍵詞抽取。

說明：本項目剛開始接觸，因此效果層面需迭代調優。

2. 技術選型

(1) 詞典

1) 基於HanLP項目提供的詞典數據，具體可參見HanLP的github。

2) 考慮到影視的垂直領域特性，引入騰訊的嵌入的漢語詞，參考該地址。

(2) 分詞

1) LTP分詞服務：基於Docker Swarm部署多副本集服務，通過HTTP協議請求，獲取分詞結果(部署方法可百度); 也可以直接在本地加載，放在內存中調用，效率更高(未嘗試)

2) AC雙數組：基於AC雙數組，采用最長匹配串，采用HanLP中的AC雙數組分詞器

(3) 抽取

1) 經典的TF-IDF：基於詞頻統計實現

2) TextRank：借鑒於PageRank算法，基於HanLP提供的接口

3) 互信息：基於HanLP提供的接口

3. 實現代碼

(1) 代碼結構

1) 代碼將分詞服務進行函數封裝，基於不同的名稱，執行名稱指定的分詞

2) TextRank、互信息、LTP、AC雙數組等提取出分詞或短語，最后均通過TF-IDF進行統計計算

(2) 整體代碼

1) 主體代碼：細節層面與下載的原始評論數據結構有關，因此無需過多關注，只需關注下主體流程即可

def extractFilmKeyWords(algorithm: String): Unit ={
    // 測試
	println(HanLPSpliter.getInstance.seg("如何看待《戰狼2》中的愛國情懷？"))

    val sc = new SparkContext(new SparkConf().setAppName("extractFileKeyWords").set("spark.driver.maxResultSize", "3g"))

    val baseDir = "/work/ws/video/parse/key_word"

    import scala.collection.JavaConversions._
    def extractComments(sc: SparkContext, inputInfo: (String, String)): RDD[(String, List[String])] = {
      sc.textFile(s"$baseDir/data/${inputInfo._2}")
        .map(data => {
          val json = JSONObjectEx.fromObject(data.trim)
          if(null == json) ("", List())
          else{
            val id = json.getStringByKeys("_id")
            val comments: List[String] = json.getArrayInfo("comments", "review").toList
            val reviews: List[String] = json.getArrayInfo("reviews", "review").toList
            val titles: List[String] = json.getArrayInfo("reviews", "title").toList
            val texts = (comments ::: reviews ::: titles).filter(f => !CleanUtils.isEmpty(f))
            (IdBuilder.getSourceKey(inputInfo._1, id), texts)
          }
        })
    }

    // 廣播停用詞
    val filterWordRdd = sc.broadcast(sc.textFile(s"$baseDir/data/stopwords.txt").map(_.trim).distinct().collect().toList)

    def formatOutput(infos: List[(Int, String)]): String ={
      infos.map(info => {
        val json = new JSONObject()
        json.put("status", info._1)
        try{
          json.put("res", info._2)
        } catch {
          case _ => json.put("res", "[]")
        }
        json.toString.replaceAll("[\\s]+", "")
      }).mkString(" | ")
    }

    def genContArray(words: List[String]): JSONArray ={
      val arr = new JSONArray()
      words.map(f => {
        val json = new JSONObject()
        json.put("cont", f)
        arr.put(json)
      })
      arr
    }

	// 基於LTP分詞服務
    def splitWordByLTP(texts: List[String]): List[(Int, String)] ={
      texts.map(f => {
        val url = "http://dev.content_ltp.research.com/ltp"
        val params = new util.HashMap[String, String]()
        params.put("s", f)
        params.put("f", "json")
        params.put("t", "ner")
        // 調用LTP分詞服務
        val result = HttpPostUtil.httpPostRetry(url, params).replaceAll("[\\s]+", "")
        if (CleanUtils.isEmpty(result)) (0, f) else {
          val resultArr = new JSONArray()

          val jsonArr = try { JSONArray.fromString(result) } catch { case _ => null}
          if (null != jsonArr && 0 < jsonArr.length()) {
            for (i <- 0 until jsonArr.getJSONArray(0).length()) {
              val subJsonArr = jsonArr.getJSONArray(0).getJSONArray(i)
              for (j <- 0 until subJsonArr.length()) {
                val subJson = subJsonArr.getJSONObject(j)
                if(!filterWordRdd.value.contains(subJson.getString("cont"))){
                  resultArr.put(subJson)
                }
              }
            }
          }
          if(resultArr.length() > 0) (1, resultArr.toString) else (0, f)
        }
      })
    }

	// 基於AC雙數組搭建的分詞服務
    def splitWordByAcDoubleTreeServer(texts: List[String]): List[(Int, String)] ={
      texts.map(f => {
        val splitResults = SplitQueryHelper.splitQueryText(f)
          .filter(f => !CleanUtils.isEmpty(f) && !filterWordRdd.value.contains(f.toLowerCase)).toList
        if (0 == splitResults.size) (0, f) else (1, genContArray(splitResults).toString)
      })
    }

	// 內存加載AC雙數組
    def splitWordByAcDoubleTree(texts: List[String]): List[(Int, String)] ={
      texts.map(f => {
        val splitResults =  HanLPSpliter.getInstance().seg(f)
          .filter(f => !CleanUtils.isEmpty(f) && !filterWordRdd.value.contains(f.toLowerCase)).toList
        if (0 == splitResults.size) (0, f) else (1, genContArray(splitResults).toString)
      })
    }

	// TextRank
    def splitWordByTextRank(texts: List[String]): List[(Int, String)] ={
      texts.map(f => {
        val splitResults = HanLP.extractKeyword(f, 100)
          .filter(f => !CleanUtils.isEmpty(f) && !filterWordRdd.value.contains(f.toLowerCase)).toList
        if (0 == splitResults.size) (0, f) else {
          val arr = genContArray(splitResults)
          if(0 == arr.length()) (0, f) else (1, arr.toString)
        }
      })
    }

	// 互信息
    def splitWordByMutualInfo(texts: List[String]): List[(Int, String)] ={
      texts.map(f => {
        val splitResults = HanLP.extractPhrase(f, 50)
          .filter(f => !CleanUtils.isEmpty(f) && !filterWordRdd.value.contains(f.toLowerCase)).toList
        if (0 == splitResults.size) (0, f) else {
          val arr = genContArray(splitResults)
          if(0 == arr.length()) (0, f) else (1, arr.toString)
        }
      })
    }

    // 提取分詞信息
    val unionInputRdd = sc.union(
	  extractComments(sc, SourceType.DB -> "db_review.json"),
      extractComments(sc, SourceType.MY -> "my_review.json"),
      extractComments(sc, SourceType.MT -> "mt_review.json"))
      .filter(_._2.nonEmpty)

    unionInputRdd.cache()

    unionInputRdd.map(data => {
      val splitResults = algorithm match {
        case "ltp" => splitWordByLTP(data._2)
        case "acServer" => splitWordByAcDoubleTreeServer(data._2)
        case "ac" => splitWordByAcDoubleTree(data._2)
        case "textRank" => splitWordByTextRank(data._2)
        case "mutualInfo" => splitWordByMutualInfo(data._2)
      }

      val output = formatOutput(splitResults)
      s"${data._1}\t$output"
    }).saveAsTextFile(HDFSFileUtil.clean(s"$baseDir/result/wordSplit/$algorithm"))

    val splitRDD = sc.textFile(s"$baseDir/result/wordSplit/$algorithm/part*", 30)
      .flatMap(data => {
        if(data.split("\\t").length < 2) None
        else{
          val sourceKey = data.split("\\t")(0)
          val words = data.split("\\t")(1).split(" \\| ").flatMap(f => {
            val json = JSONObjectEx.fromObject(f.trim)
            if (null != json && "1".equals(json.getStringByKeys("status"))) {
              val jsonArr = try { JSONArray.fromString(json.getStringByKeys("res")) } catch { case _ => null }
              var result: List[(String, String)] = List()
              if (jsonArr != null) {
                for (j <- 0 until jsonArr.length()) {
                  val json = jsonArr.getJSONObject(j)
                  val cont = json.getString("cont")
                  result ::= (cont, cont)
                }
              }
              result.reverse
            } else None
          }).toList
          Some((sourceKey, words))
        }
      }).filter(_._2.nonEmpty)

    splitRDD.cache()

    val totalFilms = splitRDD.count()

    val idfRdd = splitRDD.flatMap(result => {
      result._2.map(_._1).distinct.map((_, 1))
    }).groupByKey().filter(f => f._2.size > 1).map(f => (f._1, Math.log(totalFilms * 1.0 / (f._2.sum + 1))))

    idfRdd.cache()
    idfRdd.map(f => s"${f._1}\t${f._2}").saveAsTextFile(HDFSFileUtil.clean(s"$baseDir/result/idf/$algorithm"))

    val idfMap = sc.broadcast(idfRdd.collectAsMap())
    // 計算TF
    val tfRdd = splitRDD.map(result => {
      val totalWords = result._2.size
      val keyWords = result._2.groupBy(_._1)
        .map(f => {
          val word = f._1
          val tf = f._2.size * 1.0 / totalWords
          (tf * idfMap.value.getOrElse(word, 0D), word)
        }).toList.sortBy(_._1).reverse.filter(_._2.trim.length > 1).take(50)
      (result._1, keyWords)
    })

    tfRdd.cache()
    tfRdd.map(f => {
      val json = new JSONObject()
      json.put("_id", f._1)

      val arr = new JSONArray()
      for (keyWord <- f._2) {
        val subJson = new JSONObject()
        subJson.put("score", keyWord._1)
        subJson.put("word", keyWord._2)
        arr.put(subJson)
      }
      json.put("keyWords", arr)
      json.toString
    }).saveAsTextFile(HDFSFileUtil.clean(s"$baseDir/result/keyword/$algorithm/withScore"))

    tfRdd.map(f => s"${f._1}\t${f._2.map(_._2).toList.mkString(",")}")
      .saveAsTextFile(HDFSFileUtil.clean(s"$baseDir/result/keyword/$algorithm/noScore"))

    tfRdd.unpersist()

    splitRDD.unpersist()
    idfMap.unpersist()
    idfRdd.unpersist()

    unionInputRdd.unpersist()
    filterWordRdd.unpersist()
    sc.stop()
  }

2) 基於HanLP提供的AC雙數組封裝

import com.google.common.collect.Lists;
import com.hankcs.hanlp.HanLP;
import com.hankcs.hanlp.seg.Segment;
import com.hankcs.hanlp.seg.common.Term;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.List;

public class HanLPSpliter implements Serializable{
    private static Logger logger = LoggerFactory.getLogger(Act.class);

    private static HanLPSpliter instance = null;

    private static Segment segment = null;

    private static final String PATH = "conf/tencent_word_act.txt";

    public static HanLPSpliter getInstance() {
        if(null == instance){
            instance = new HanLPSpliter();
        }
        return instance;
    }

    public HanLPSpliter(){
        this.init();
    }

    public void init(){
        initSegment();
    }

    public void initSegment(){
        if(null == segment){
            addDict();
            HanLP.Config.IOAdapter = new HadoopFileIOAdapter();
            segment = HanLP.newSegment("dat");
            segment.enablePartOfSpeechTagging(true);
            segment.enableCustomDictionaryForcing(true);
        }
    }

    public List<String> seg(String text){
        if(null == segment){
            initSegment();
        }

        List<Term> terms = segment.seg(text);
        List<String> results = Lists.newArrayList();
        for(Term term : terms){
            results.add(term.word);
        }
        return results;
    }
}

3) HanLP加載HDFS中的自定義詞典

import com.hankcs.hanlp.corpus.io.IIOAdapter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;

public class HadoopFileIOAdapter implements IIOAdapter{
    @Override
    public InputStream open(String path) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(path), conf);
        return fs.open(new Path(path));
    }

    @Override
    public OutputStream create(String path) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(path), conf);
        OutputStream out = fs.create(new Path(path));
        return out;
    }
}

4. 采坑總結

(1) Spark中實現HanLP自定義詞典的加載

由於引入騰訊的嵌入詞，因此使用HanLP的自定義詞典功能，參考的方法如下：

a. 《基於hanLP的中文分詞詳解-MapReduce實現&自定義詞典文件》，該方法適用於自定義詞典的數量較少的情況，如果詞典量較大，如騰訊嵌入詞820W+，理論上jar包較為臃腫

b. 《Spark中使用HanLP分詞》，該方法的好處在於無需手工構件詞典的bin文件，操作簡單

切記：如果想讓自定義詞典生效，需先將data/dictionary/custom中的bin文件刪除。通過HanLP源碼得知，如果存在bin文件，則直接加載該bin文件，否則會將custom中用戶自定義的詞典重新加載，在指定的環境中(如本地或HDFS)中自動生成bin文件。

騰訊820W詞典，基於HanLP生成bin文件的時間大概為30分鍾。

(2) Spark異常

Spark執行過程中的異常信息：

1) 異常1

a. 異常信息：

Job aborted due to stage failure: Total size of serialized results of 3979 tasks (1024.2 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)

b. 解決：通過設置spark.driver.maxResultSize=4G，參考：《Spark排錯與優化》

2) 異常2

a. 異常信息：java.lang.OutOfMemoryError: Java heap space

b. 解決：參考https://blog.csdn.net/guohecang/article/details/52088117

 

如有問題，請留言回復！