devbox
小丑西瓜9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

spark应用（三）文本分类_spark的中文文本分类

作者：小丑西瓜9 | 2024-05-13 17:22:47

赞

踩

spark的中文文本分类

一、特征提取

1、什么是特征提取？
     对某一模式的组测量值进行变换，以突出该模式具有代表性特征的一种方法（百度百科）。或者参考多方面的解释 ： 
http://www.igi-global.com/dictionary/feature-extraction/10960

特征提取简单来说就是从一堆数据中，提取数据，并变成我们熟悉的数据形式（比如从图片中提取像素点，并变成RGB数字，或者把文档变成我们熟悉的向量空间）

2、TF-IDF
     TF-IDF是在文本挖掘中广泛使用的特征向量方法，以反映term（术语）对语料库中的文档的重要性。term的频率是term在文档中出现的次数除以总文档的占比。而文档的频率是包含這个term占除以总体文档的占比。如果单从TF（term频率），很容易出现强调出现很频率的term,百比如英文中，'a',‘the’,'of'。IDF（反向文档频率）就说明该term会不会使用很平凡。
TF-IDF计算公式如下：

对于更多细节，可以参考之前写的：

http://blog.csdn.net/legotime/article/details/51836028
在实际中，往往IDF的分子和分母都会加1

3、TF计算
     TF计算Spark提供了HashingTF和CountVectorizer
HashingTF
       HashingTF方法用的是hash trick(feature hash).而spark中用的是MurmurHash3 （ https://en.wikipedia.org/wiki/MurmurHash）的算法。为了让大家有一个更深层次的认识。现在来说说什么是hash trick。如下三句话 
  1. John likes to watch movies.
  2. Mary likes movies too.
  3. John also likes football.
转换为：

Term Index
John1
likes2
to3
watch4
movies5
Mary6
too7
also8
football9
（就是从John开始，从左到右，从上到下。重复的就pass）,那么三个句子的term-document矩阵如下：

现在用Java来实现：

  1. /**
  2. *
  3. * @param file 文件位置
  4. * @return
  5. * @throws IOException
  6. */
  7. public static ArrayList<int[]> txt2num(String file) throws IOException {
  8.     BufferedReader br = new BufferedReader(new FileReader(file));
  9.     String s;
  10.     StringBuilder sb = new StringBuilder();
  11.     ArrayList<String> strArr = new ArrayList<String>();
  12.     while ((s=br.readLine()) != null){
  13.         String tmp = s.split("\\.")[0];
  14.         strArr.add(tmp);
  15.         sb.append(tmp+" ");
  16.     }
  17.     String[] split = sb.toString().split(" ");
  18.     TreeSet<String> strHashSet = new TreeSet<>();
  19.     for (String s1 : split) {
  20.         strHashSet.add(s1);
  21.     }
  22.     ArrayList<int[]> txt2Matrix = new ArrayList<int[]>();
  23.     System.out.println(Arrays.toString(strHashSet.toArray()));
  24.     //填入数据
  25.     for (String s1 : strArr) {
  26.         int[] txt2IntVec = new int[strHashSet.size()];
  27.         String[] ss = s1.split(" ");
  28.         ArrayList<String > strs = new ArrayList<String>();
  29.         for (String s2 : ss) {
  30.             strs.add(s2);
  31.         }
  32.         System.out.println(Arrays.toString(ss));
  33.         for (int i = 0; i < txt2IntVec.length; i++) {
  34.             txt2IntVec[i] = strs.contains(strHashSet.toArray()[i]) ? 1 : 0;
  35.         }
  36.         System.out.println(Arrays.toString(txt2IntVec));
  37.         txt2Matrix.add(txt2IntVec);
  38.     }
  39.     return txt2Matrix;
  40. }

可以看看MLlib下面的MurmurHash3

  1. public final class Murmur3_x86_32 {
  2.   private static final int C1 = 0xcc9e2d51;
  3.   private static final int C2 = 0x1b873593;
  4.  
  5.   private final int seed;
  6.  
  7.   public Murmur3_x86_32(int seed) {
  8.     this.seed = seed;
  9.   }
  10.  
  11.   @Override
  12.   public String toString() {
  13.     return "Murmur3_32(seed=" + seed + ")";
  14.   }
  15.  
  16.   public int hashInt(int input) {
  17.     return hashInt(input, seed);
  18.   }
  19.  
  20.   public static int hashInt(int input, int seed) {
  21.     int k1 = mixK1(input);
  22.     int h1 = mixH1(seed, k1);
  23.  
  24.     return fmix(h1, 4);
  25.   }
  26.  
  27.   public int hashUnsafeWords(Object base, long offset, int lengthInBytes) {
  28.     return hashUnsafeWords(base, offset, lengthInBytes, seed);
  29.   }
  30.  
  31.   public static int hashUnsafeWords(Object base, long offset, int lengthInBytes, int seed) {
  32.     // This is based on Guava's `Murmur32_Hasher.processRemaining(ByteBuffer)` method.
  33.     assert (lengthInBytes % 8 == 0): "lengthInBytes must be a multiple of 8 (word-aligned)";
  34.     int h1 = hashBytesByInt(base, offset, lengthInBytes, seed);
  35.     return fmix(h1, lengthInBytes);
  36.   }
  37.  
  38.   public static int hashUnsafeBytes(Object base, long offset, int lengthInBytes, int seed) {
  39.     assert (lengthInBytes >= 0): "lengthInBytes cannot be negative";
  40.     int lengthAligned = lengthInBytes - lengthInBytes % 4;
  41.     int h1 = hashBytesByInt(base, offset, lengthAligned, seed);
  42.     for (int i = lengthAligned; i < lengthInBytes; i++) {
  43.       int halfWord = Platform.getByte(base, offset + i);
  44.       int k1 = mixK1(halfWord);
  45.       h1 = mixH1(h1, k1);
  46.     }
  47.     return fmix(h1, lengthInBytes);
  48.   }
  49.  
  50.   private static int hashBytesByInt(Object base, long offset, int lengthInBytes, int seed) {
  51.     assert (lengthInBytes % 4 == 0);
  52.     int h1 = seed;
  53.     for (int i = 0; i < lengthInBytes; i += 4) {
  54.       int halfWord = Platform.getInt(base, offset + i);
  55.       int k1 = mixK1(halfWord);
  56.       h1 = mixH1(h1, k1);
  57.     }
  58.     return h1;
  59.   }
  60.  
  61.   public int hashLong(long input) {
  62.     return hashLong(input, seed);
  63.   }
  64.  
  65.   public static int hashLong(long input, int seed) {
  66.     int low = (int) input;
  67.     int high = (int) (input >>> 32);
  68.  
  69.     int k1 = mixK1(low);
  70.     int h1 = mixH1(seed, k1);
  71.  
  72.     k1 = mixK1(high);
  73.     h1 = mixH1(h1, k1);
  74.  
  75.     return fmix(h1, 8);
  76.   }
  77.  
  78.   private static int mixK1(int k1) {
  79.     k1 *= C1;
  80.     k1 = Integer.rotateLeft(k1, 15);
  81.     k1 *= C2;
  82.     return k1;
  83.   }
  84.  
  85.   private static int mixH1(int h1, int k1) {
  86.     h1 ^= k1;
  87.     h1 = Integer.rotateLeft(h1, 13);
  88.     h1 = h1 * 5 + 0xe6546b64;
  89.     return h1;
  90.   }
  91.  
  92.   // Finalization mix - force all bits of a hash block to avalanche
  93.   private static int fmix(int h1, int length) {
  94.     h1 ^= length;
  95.     h1 ^= h1 >>> 16;
  96.     h1 *= 0x85ebca6b;
  97.     h1 ^= h1 >>> 13;
  98.     h1 *= 0xc2b2ae35;
  99.     h1 ^= h1 >>> 16;
  100.     return h1;
  101.   }
  102. }
CountVectorizer
  比较简单，可以看看如下：和hash trick区别在于，它是对某个term进行了计算。


二、文本分类

数据

http://qwone.com/~jason/20Newsgroups/

上面已经有数据的属性说明，
选用的数据集是：

导入HDFS如下：



数据预处理
     把数据转换为如下格式： 
case class LabeledText(item:String,label:Double,doc:String)
其中： 
  1. item:文件名字（类名）
  2. label:标签
  3. doc:从整个文本中提取的单词或者字母
处理程序如下： 

  1. object NewClassifier {
  2.  
  3.   def listSonRoute(path: String): Seq[String] ={
  4.  
  5.     val conf = new Configuration()
  6.     val fs = new Path(path).getFileSystem(conf)
  7.     val status = fs.listFiles(new Path(path),true)
  8.     var res: List[String] = Nil
  9.     while (status.hasNext){
  10.       res = res++Seq(status.next().getPath.toString)
  11.     }
  12.     res
  13.   }
  14.   /**
  15.     * 提取英文单词或者字母
  16.     * @param content
  17.     * @return
  18.     */
  19.   def splitStr(content: String): List[String] =("[A-Za-z]+$".r findAllIn content).toList
  20.  
  21.   def rdd2Str(sc:SparkContext,path:String)= {
  22.     val rdd = sc.textFile(path)
  23.     val myAccumulator = sc.accumulator[String](" ")(StringAccumulatorParam)
  24.     rdd.foreach{ part=> splitStr(part).foreach{ word =>
  25.       myAccumulator.add(word.toLowerCase)
  26.     }}
  27.     myAccumulator.value
  28.   }
  29.   def getDataFromHDFS(sc:SparkContext,path:String): DataFrame ={
  30.     val sqlContext = new SQLContext(sc)
  31.     import sqlContext.implicits._
  32.     listSonRoute(path).map(
  33.       part =>
  34.       LabeledText(part.split("/").apply(8),new Random(100).nextInt(),rdd2Str(sc,part))
  35.     ).toDF()
  36.   }
  37.   case class LabeledText(item:String,label:Double,doc:String)
  38.   def main(args: Array[String]) {
  39.     val conf = new SparkConf().setAppName("new Classifier").setMaster("local")
  40.       .set("spark.storage.memoryFraction", "0.1")
  41.  
  42.     val sc = new SparkContext(conf)
  43.     // rawData to parquet
  44.     val testPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/20news-bydate-test"
  45.     val trainPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/20news-bydate-train/"
  46.     val testDF = getDataFromHDFS(sc,testPath)
  47.     val trainDF = getDataFromHDFS(sc,trainPath)
  48.  
  49.     testDF.write.save("hdfs://master:9000/data/studySet/textMining/20news-bydate/test")
  50.     trainDF.write.save("hdfs://master:9000/data/studySet/textMining/20news-bydate/train")
  51.   }
  52. }
  53. object StringAccumulatorParam extends AccumulatorParam[String] {
  54.   override def addInPlace(r1: String, r2: String): String = add(r1,r2)
  55.  
  56.   /**
  57.     * 初始化
  58.     * @param initialValue　初始值
  59.     * @return
  60.     */
  61.   override def zero(initialValue: String): String = ""
  62.   def  add(v1:String,v2:String)={
  63.     assert((!v1.isEmpty)|| (!v2.isEmpty))
  64.     v1+v2+" "
  65.   }
  66. }

贝叶斯分类
     下面结合pipeline的处理方式用贝叶斯对文本进行分类，程序如下： 
  1. package txtMIning
  2.  
  3. import org.apache.hadoop.conf.Configuration
  4. import org.apache.hadoop.fs._
  5. import org.apache.spark.ml.Pipeline
  6. import org.apache.spark.ml.classification.{DecisionTreeClassifier, NaiveBayes}
  7. import org.apache.spark.ml.feature.{HashingTF, RegexTokenizer}
  8. import org.apache.spark.sql.{DataFrame, SQLContext}
  9. import org.apache.spark.{AccumulatorParam, SparkConf, SparkContext}
  10.  
  11. import scala.util.Random
  12.  
  13. /**
  14.   * 新闻分类
  15.   */
  16. object NewClassifier {
  17.  
  18.   def listSonRoute(path: String): Seq[String] ={
  19.  
  20.     val conf = new Configuration()
  21.     val fs = new Path(path).getFileSystem(conf)
  22.     val status = fs.listFiles(new Path(path),true)
  23.     var res: List[String] = Nil
  24.     while (status.hasNext){
  25.       res = res++Seq(status.next().getPath.toString)
  26.     }
  27.     res
  28.   }
  29.   /**
  30.     * 提取英文单词或者字母
  31.     * @param content
  32.     * @return
  33.     */
  34.   def splitStr(content: String): List[String] =("[A-Za-z]+$".r findAllIn content).toList
  35.  
  36.   def rdd2Str(sc:SparkContext,path:String)= {
  37.     val rdd = sc.textFile(path)
  38.     val myAccumulator = sc.accumulator[String](" ")(StringAccumulatorParam)
  39.     rdd.foreach{ part=> splitStr(part).foreach{ word =>
  40.       myAccumulator.add(word.toLowerCase)
  41.     }}
  42.     myAccumulator.value
  43.   }
  44.   def getDataFromHDFS(sc:SparkContext,path:String): DataFrame ={
  45.     val sqlContext = new SQLContext(sc)
  46.     import sqlContext.implicits._
  47.     listSonRoute(path).map(
  48.       part =>
  49.       LabeledText(part.split("/").apply(8),new Random(100).nextInt(),rdd2Str(sc,part))
  50.     ).toDF()
  51.   }
  52.   def readParquetFile(sc:SparkContext,path:String)={
  53.     val sqlContext = new SQLContext(sc)
  54.     sqlContext.read.parquet(path).toDF()
  55.   }
  56.  
  57.  
  58.  
  59.  
  60.   case class LabeledText(item:String,label:Double,doc:String)
  61.   /**
  62.     *
  63.     *
  64.     * @param args
  65.     */
  66.   def main(args: Array[String]) {
  67.     val conf = new SparkConf().setAppName("new Classifier").setMaster("local")
  68.       .set("spark.storage.memoryFraction", "0.1")
  69.  
  70.     val sc = new SparkContext(conf)
  71.  
  72.  
  73. //    // rawData to parquet
  74. //    val testPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/20news-bydate-test"
  75. //    val trainPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/20news-bydate-train/"
  76. //    val testDF = getDataFromHDFS(sc,testPath)
  77. //    val trainDF = getDataFromHDFS(sc,trainPath)
  78.  
  79. //    testDF.write.save("hdfs://master:9000/data/studySet/textMining/20news-bydate/test")
  80. //    trainDF.write.save("hdfs://master:9000/data/studySet/textMining/20news-bydate/train")
  81.  
  82.     //数据增加label(而且这个label必须是Double类型)
  83.     val testParquetPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/test/*"
  84.     val trainParquetPath = "hdfs://master:9000/data/studySet/textMining/20news-bydate/train/*"
  85.  
  86.  
  87.     val testDF: DataFrame = readParquetFile(sc,testParquetPath)//.sample(withReplacement = true,0.002)
  88.     val trainDF: DataFrame = readParquetFile(sc,trainParquetPath)//.sample(withReplacement = true,0.002)
  89.  
  90. //    val pre = readParquetFile(sc,"hdfs://master:9000/data/studySet/textMining/20news-bydate/prediction2/*")
  91. //    pre.show(200)
  92. //    pre.toJavaRDD.saveAsTextFile("hdfs://master:9000/data/studySet/textMining/20news-bydate/prediction3")
  93.     //testDF.foreach(println)
  94.     //val trainDF: DataFrame = readParquetFile(sc,trainParquetPath)
  95.  
  96.     //trainDF.show()
  97. //    testDF.show(5)
  98.  
  99.  
  100.  
  101.  
  102.     //testDF.take(1).foreach(println)
  103.     //[alt.atheism, answers t na translator had but determine kaflowitz  ]
  104.  
  105.  
  106.     //use the pipeline
  107.     val tokenizer = new RegexTokenizer()
  108.       .setInputCol("doc")
  109.       .setOutputCol("words")
  110.  
  111.     val hashingTF = new HashingTF()
  112.       .setInputCol(tokenizer.getOutputCol)
  113.       .setOutputCol("features")
  114. //      .setNumFeatures(100000)
  115.  
  116.     val naiveBayes = new NaiveBayes()
  117.       .setPredictionCol("prediction")
  118.  
  119.     val decisionTree = new DecisionTreeClassifier()
  120.       .setPredictionCol("prediction")
  121.     val pipeline = new Pipeline().setStages(Array(tokenizer,hashingTF,decisionTree))
  122.  
  123.  
  124. //    //???????
  125.     val model = pipeline.fit(testDF)
  126.     //println(model.explainParams())
  127.     val trainPredictions = model.transform(trainDF)
  128.     trainPredictions.show(50)
  129.     //trainPredictions.write.save("hdfs://master:9000/data/studySet/textMining/20news-bydate/prediction2")
  130.  
  131.  
  132.   }
  133.  
  134. }
  135. object StringAccumulatorParam extends AccumulatorParam[String] {
  136.   override def addInPlace(r1: String, r2: String): String = add(r1,r2)
  137.  
  138.   /**
  139.     * 初始化
  140.     * @param initialValue　初始值
  141.     * @return
  142.     */
  143.   override def zero(initialValue: String): String = ""
  144.   def  add(v1:String,v2:String)={
  145.     assert((!v1.isEmpty)|| (!v2.isEmpty))
  146.     v1+v2+" "
  147.   }
  148. }

结果如下：















声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/564651
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号