scala - 增加 MinHashLSH 中的哈希表，降低准确性和 f1

Question

我已经使用 MinHashLSH 和approximateSimilarityJoin 以及 Scala 和 Spark 2.4 来查找网络之间的边缘。基于文档相似度的链接预测。我的问题是，当我增加 MinHashLSH 中的哈希表时，我的准确性和 F1 分数正在下降。我已经为这个算法阅读的所有内容都表明我有一个问题。

我尝试了不同数量的哈希表，并提供了不同数量的 Jaccard 相似度阈值，但我遇到了同样的问题，准确度正在迅速下降。我也尝试过对我的数据集进行不同的采样，但没有任何改变。我的工作流程是这样进行的：我正在连接我的数据框的所有文本列，其中包括标题、作者、期刊和摘要，接下来我将连接的列标记为单词。然后我使用 CountVectorizer 将这个“词袋”转换为向量。接下来，我在 MinHashLSH 中为该列提供一些哈希表，最后我正在做一个近似相似性连接来查找低于我给定阈值的类似“论文”。我的实现如下。

import org.apache.spark.ml.feature._
import org.apache.spark.ml.linalg._
import UnsupervisedLinkPrediction.BroutForce.join
import org.apache.log4j.{Level, Logger}
import org.apache.spark.ml.Pipeline
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, udf, when}
import org.apache.spark.sql.types._


object lsh {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR) // show only errors

//    val cores=args(0).toInt
//    val partitions=args(1).toInt
//    val hashTables=args(2).toInt
//    val limit = args(3).toInt
//    val threshold = args(4).toDouble

    val cores="*"
    val partitions=1
    val hashTables=16
    val limit = 1000
    val jaccardDistance = 0.89

    val master = "local["+cores+"]"

    val ss = SparkSession.builder().master(master).appName("MinHashLSH").getOrCreate()
    val sc = ss.sparkContext

    val inputFile = "resources/data/node_information.csv"

    println("reading from input file: " + inputFile)
    println

    val schemaStruct = StructType(
      StructField("id", IntegerType) ::
        StructField("pubYear", StringType) ::
        StructField("title", StringType) ::
        StructField("authors", StringType) ::
        StructField("journal", StringType) ::
        StructField("abstract", StringType) :: Nil
    )

    // Read the contents of the csv file in a dataframe. The csv file contains a header.
    //    var papers = ss.read.option("header", "false").schema(schemaStruct).csv(inputFile).limit(limit).cache()

    var papers = ss.read.option("header", "false").schema(schemaStruct).csv(inputFile).limit(limit).cache()

    papers.repartition(partitions)
    println("papers.rdd.getNumPartitions"+papers.rdd.getNumPartitions)

    import ss.implicits._
    // Read the original graph edges, ground trouth
    val originalGraphDF = sc.textFile("resources/data/Cit-HepTh.txt").map(line => {
      val fields = line.split("\t")
      (fields(0), fields(1))
    }).toDF("nodeA_id", "nodeB_id").cache()

    val originalGraphCount = originalGraphDF.count()

    println("Ground truth count: " + originalGraphCount )

    val nullAuthor = ""
    val nullJournal = ""
    val nullAbstract = ""

    papers = papers.na.fill(nullAuthor, Seq("authors"))
    papers = papers.na.fill(nullJournal, Seq("journal"))
    papers = papers.na.fill(nullAbstract, Seq("abstract"))

    papers = papers.withColumn("nonNullAbstract", when(col("abstract") === nullAbstract, col("title")).otherwise(col("abstract")))
    papers = papers.drop("abstract").withColumnRenamed("nonNullAbstract", "abstract")
    papers.show(false)

        val filteredGt= originalGraphDF.as("g").join(papers.as("p"),(
          $"g.nodeA_id" ===$"p.id") || ($"g.nodeB_id" ===$"p.id")
        ).select("g.nodeA_id","g.nodeB_id").distinct().cache()

    filteredGt.show()

    val filteredGtCount = filteredGt.count()
    println("Filtered GroundTruth count: "+ filteredGtCount)

    //TOKENIZE

    val tokPubYear = new Tokenizer().setInputCol("pubYear").setOutputCol("pubYear_words")
    val tokTitle = new Tokenizer().setInputCol("title").setOutputCol("title_words")
    val tokAuthors = new RegexTokenizer().setInputCol("authors").setOutputCol("authors_words").setPattern(",")
    val tokJournal = new Tokenizer().setInputCol("journal").setOutputCol("journal_words")
    val tokAbstract = new Tokenizer().setInputCol("abstract").setOutputCol("abstract_words")

    println("Setting pipeline stages...")
    val stages = Array(
      tokPubYear, tokTitle, tokAuthors, tokJournal, tokAbstract
      //      rTitle, rAuthors, rJournal, rAbstract
    )

    val pipeline = new Pipeline()
    pipeline.setStages(stages)

    println("Transforming dataframe\n")
    val model = pipeline.fit(papers)
    papers = model.transform(papers)

    println(papers.count())
    papers.show(false)
    papers.printSchema()

    val udf_join_cols = udf(join(_: Seq[String], _: Seq[String], _: Seq[String], _: Seq[String], _: Seq[String]))

    val joinedDf = papers.withColumn(
      "paper_data",
      udf_join_cols(
        papers("pubYear_words"),
        papers("title_words"),
        papers("authors_words"),
        papers("journal_words"),
        papers("abstract_words")
      )
    ).select("id", "paper_data").cache()

    joinedDf.show(5,false)

    val vocabSize = 1000000
    val cvModel: CountVectorizerModel = new CountVectorizer().setInputCol("paper_data").setOutputCol("features").setVocabSize(vocabSize).setMinDF(10).fit(joinedDf)
    val isNoneZeroVector = udf({v: Vector => v.numNonzeros > 0}, DataTypes.BooleanType)
    val vectorizedDf = cvModel.transform(joinedDf).filter(isNoneZeroVector(col("features"))).select(col("id"), col("features"))
    vectorizedDf.show()

    val mh = new MinHashLSH().setNumHashTables(hashTables)
      .setInputCol("features").setOutputCol("hashValues")
    val mhModel = mh.fit(vectorizedDf)

    mhModel.transform(vectorizedDf).show()

    vectorizedDf.createOrReplaceTempView("vecDf")

    println("MinHashLSH.getHashTables: "+mh.getNumHashTables)

    val dfA = ss.sqlContext.sql("select id as nodeA_id, features from vecDf").cache()
    dfA.show(false)
    val dfB = ss.sqlContext.sql("select id as nodeB_id, features from vecDf").cache()
    dfB.show(false)

    val predictionsDF = mhModel.approxSimilarityJoin(dfA, dfB, jaccardDistance, "JaccardDistance").cache()

    println("Predictions:")
    val predictionsCount = predictionsDF.count()
    predictionsDF.show()
    println("Predictions count: "+predictionsCount)

        predictionsDF.createOrReplaceTempView("predictions")

        val pairs = ss.sqlContext.sql("select datasetA.nodeA_id, datasetB.nodeB_id, JaccardDistance from predictions").cache()
        pairs.show(false)

        val totalPredictions = pairs.count()

        println("Properties:\n")
        println("Threshold: "+threshold+"\n")
        println("Hahs tables: "+hashTables+"\n")
        println("Ground truth: "+filteredGtCount)
        println("Total edges found: "+totalPredictions +" \n")


        println("EVALUATION PROCESS STARTS\n")
        println("Calculating true positives...\n")

        val truePositives = filteredGt.as("g").join(pairs.as("p"),
          ($"g.nodeA_id" === $"p.nodeA_id" && $"g.nodeB_id" === $"p.nodeB_id") || ($"g.nodeA_id" === $"p.nodeB_id" && $"g.nodeB_id" === $"p.nodeA_id")
        ).cache().count()

       println("True Positives: "+truePositives+"\n")

        println("Calculating false positives...\n")

        val falsePositives = predictionsCount - truePositives

        println("False Positives: "+falsePositives+"\n")

        println("Calculating true negatives...\n")
        val pairsPerTwoCount = (limit *(limit - 1)) / 2

        val trueNegatives = (pairsPerTwoCount - truePositives) - falsePositives
        println("True Negatives: "+trueNegatives+"\n")

        val falseNegatives = filteredGtCount - truePositives

        println("False Negatives: "+falseNegatives)

        val truePN = (truePositives+trueNegatives).toFloat
        println("TP + TN sum: "+truePN+"\n")

        val sum = (truePN + falseNegatives+ falsePositives).toFloat
        println("TP +TN +FP+ FN sum: "+sum+"\n")

        val accuracy = (truePN/sum).toFloat
        println("Accuracy: "+accuracy+"\n")

        val precision = truePositives.toFloat / (truePositives+falsePositives).toFloat
        val recall = truePositives.toFloat/(truePositives+falseNegatives).toFloat

        val f1Score = 2*(recall*precision)/(recall+precision).toFloat
        println("F1 score: "+f1Score+"\n")

    ss.stop()

我忘了告诉你，我在一个有 40 个内核和 64g RAM 的集群中运行这段代码。请注意，近似相似连接（Spark 的实现）适用于 JACCARD DISTANCE 而不是 JACCARD INDEX。因此，我提供了 JACCARD DISTANCE 作为相似度阈值，在我的情况下，它是 jaccardDistance = 1 - 阈值。（阈值 = Jaccard 指数）。

我希望在增加哈希表时获得更高的准确性和 f1 分数。你对我的问题有任何想法吗？

提前感谢大家！

Answer 1

这里有多个可见的问题，而且可能更隐蔽，因此仅列举几个：

• LSH 并不是一个真正的分类器，并且试图将其评估为一个没有多大意义的分类器，即使您假设文本相似性在某种程度上是引用的代理（如果是这样的话）。
• 如果问题被定义为分类问题，则应将其视为多标签分类（每篇论文可以引用或被多个来源引用）而不是多类分类，因此简单的准确性没有意义。
• 即使它是一个分类并且可以这样评估，您的计算也不包括实际的否定，它们不符合approxSimilarityJoin
• 还将阈值设置为 1 将连接限制为完全匹配或哈希冲突的情况 - 因此优先选择具有更高冲突率的 LSH。

此外：

• 您采用的文本处理方法相当平庸，并且更喜欢非特定功能（请记住，您不会优化实际目标，而是优化文本相似性）。
• 这种方法，尤其是平等对待所有事物，主要丢弃集合中的大部分有用信息，但不限于时间关系。