0

学術論文のデータセットがあり、27770 の論文 (ノード) と、352807 エントリの長さの元のエッジを持つ別のファイル (グラフ ファイル) があります。minHashLSH を計算して類似のドキュメントを見つけ、2 つのノード間のリンクを予測したい! 怒鳴ると、これを scala で spark に実装する私の試みを見ることができます。私が直面している問題は、結果を評価する方法がわからないことです! 

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


val ss = SparkSession.builder().master("local[*]").appName("neighbors").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)

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")

println("Original graph edges count: " + originalGraphDF.count())
originalGraphDF.printSchema()
originalGraphDF.show(5)

val t1 = System.nanoTime // Start point of the app

val nullAuthor = "NO AUTHORS"
val nullJournal = "NO JOURNAL"
val nullAbstract = "NO ABSTRACT"

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()

papers = papers.na.drop()
val fraction = 0.1

papers = papers.sample(fraction, 12345L)
//    println(papers.count())

//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")

//REMOVE STOPWORDS

val rTitle = new StopWordsRemover().setInputCol("title_words").setOutputCol("title_words_f")
val rAuthors = new StopWordsRemover().setInputCol("authors_words").setOutputCol("authors_words_f")
val rJournal = new StopWordsRemover().setInputCol("journal_words").setOutputCol("journal_words_f")
val rAbstract = new StopWordsRemover().setInputCol("abstract_words").setOutputCol("abstract_words_f")

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")
val model = pipeline.fit(papers)
papers = model.transform(papers)

papers.show(5)

//newDf = node df
val newDf = papers.select("id", "pubYear_words", "title_words_f", "authors_words_f", "journal_words_f", "abstract_words_f")
newDf.show(5)
newDf.describe().show()

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

val joinedDf = newDf.withColumn(
  "paper_data",
  udf_join_cols(
    newDf("pubYear_words"),
    newDf("title_words_f"),
    newDf("authors_words_f"),
    newDf("journal_words_f"),
    newDf("abstract_words_f"
    )
  )
).select("id", "paper_data")

joinedDf.show(5)
joinedDf.printSchema()
println(joinedDf.count())

// Word count to vector for each wiki content
val vocabSize = 1000000
val cvModel: CountVectorizerModel = new CountVectorizer()
  .setInputCol("paper_data").setOutputCol("features").setVocabSize(vocabSize)
  .setMinDF(10).fit(joinedDf)

val vectorizedDf = cvModel.transform(joinedDf).select(col("id"), col("features"))
vectorizedDf.show()
println("Total entries: "+vectorizedDf.count())

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

mhModel.transform(vectorizedDf).show()

// Self Join
val threshold = 0.95

val predictinsDF = mhModel.approxSimilarityJoin(vectorizedDf, vectorizedDf, 1,"JaccardDistance")
  .select("datasetA.id","datasetB.id","JaccardDistance").filter("JaccardDistance >= "+threshold)
  .withColumnRenamed("datasetA.id","nodeA_id")
  .withColumnRenamed("datasetB.id","nodeB_id")

predictinsDF.show()
predictinsDF.printSchema()
println("Total edges found: "+predictinsDF.count())  }

元のグラフは、nodeAId、nodeBId の形式のファイルです。私の結果は、nodeAId、nodeBId、JaccardSimilarity の形式になっています。どちらもデータフレームです。結果を評価し、精度または F1 スコアを取得するにはどうすればよいですか?

Accuracy と F1 スコアを見つける方法を読んだので、それらを計算する関数を作成しようとしました。私のアプローチは以下のコードです。 

def getStats(spark:SparkSession,nodeDF:DataFrame, pairsDF:DataFrame, predictionsDF:DataFrame, graphDF:DataFrame): Unit ={
Logger.getLogger("org").setLevel(Level.ERROR)

import spark.implicits._
val truePositives = graphDF.as("g").join(predictionsDF.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")
).count()

val df = pairsDF.as("p").join(graphDF.as("g"),
  ($"p.nodeA_id" === $"g.nodeA_id" && $"p.nodeB_id" === $"g.nodeB_id") || ($"p.nodeA_id" === $"g.nodeB_id" && $"p.nodeB_id" === $"g.nodeA_id")
).count()
println("True Positives: "+truePositives)

val falsePositives = predictionsDF.count() - truePositives
println("False Positives: "+falsePositives)

val trueNegatives = (pairsDF.count() - graphDF.count()) - falsePositives
println("True Negatives: "+trueNegatives)

val falseNegatives = graphDF.count()-truePositives
println("False Negatives: "+falseNegatives)

val truePN = (truePositives+trueNegatives).toFloat

val sum = (truePN + falseNegatives+ falsePositives).toFloat

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

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) }

でも、走らせてみると止まらない!! 精度と F1 スコアを得るために、これを改善する方法や修正する方法がわかりません。これを行う簡単な方法はありますか？

皆さんのお陰で！

4

1 に答える 1