<commit_before>package net.sansa_stack.ml.spark.clustering

import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.clustering.{ PowerIterationClusteringModel, PowerIterationClustering }
import org.apache.spark.graphx.{ Graph, EdgeDirection }
import scala.math.BigDecimal
import org.apache.commons.math3.util.MathUtils
import org.apache.spark.sql.SparkSession

class RDFGraphPICClustering(@transient val sparkSession: SparkSession,
                            val graph: Graph[Int, Int],
                            private val k: Int,
                            private val maxIterations: Int) extends Serializable {

  def clusterRdd(): RDD[(Long, Long, Double)] = {
    SimilaritesInPIC
  }

  /*
	 * Computes different similarities function for a given graph @graph.
	 */
  def SimilaritesInPIC(): RDD[(Long, Long, Double)] = {
    /*
	 * Collect all the edges of the graph
	*/
    val edge = graph.edges.collect()
    /*
	 * Collect neighbor IDs of all the vertices
	 */
    val neighbors = graph.collectNeighborIds(EdgeDirection.Either)
    /*
	 * Collect distinct vertices of the graph
	 */
    val vertices = graph.vertices.distinct()
    /*
	 * Difference between two set of vertices, used in different similarity measures
	 */
    def difference(a: Long, b: Long): Double = {
      val ansec = neighbors.lookup(a).distinct.head.toSet
      val ansec1 = neighbors.lookup(b).distinct.head.toSet
      if (ansec.isEmpty) { return 0.0 }
      val differ = ansec.diff(ansec1)
      if (differ.isEmpty) { return 0.0 }

      differ.size.toDouble
    }

    /*
	 * Intersection of two set of vertices, used in different similarity measures
	 */
    def intersection(a: Long, b: Long): Double = {
      val inters = neighbors.lookup(a).distinct.head.toSet
      val inters1 = neighbors.lookup(b).distinct.head.toSet
      if (inters.isEmpty || inters1.isEmpty) { return 0.0 }
      val rst = inters.intersect(inters1).toArray
      if (rst.isEmpty) { return 0.0 }
      rst.size.toDouble
    }

    /*
			 * Union of two set of vertices, used in different similarity measures
			 */
    def union(a: Long, b: Long): Double = {
      val inters = neighbors.lookup(a).distinct.head.toSet
      val inters1 = neighbors.lookup(b).distinct.head.toSet
      val rst = inters.union(inters1).toArray
      if (rst.isEmpty) { return 0.0 }

      rst.size.toDouble
    }
    // Logarithm base 2 
    val LOG2 = math.log(2)
    val log2 = { x: Double => math.log(x) / LOG2 }
	  
	  
	  val vertexCount = vertices.count()
	  
	  
	  // Similarity based on Information Theory
	  
	  
     var icc = 0.0
    val logC = MathUtils.log(10.0, vertexCount.toDouble)
 
     def informationContent(a: Long): Double = {
      if (a == 0) { return 0.0 }
       1 - (MathUtils.log(10.0, a.toDouble) / logC)
 
     }
     
	  def sigmaIc(a : List[Long]) : Double ={
   var sigma = 0.0
    for(k <- 0 until a.length) yield{
      
      val s = informationContent(a(k))
      sigma = sigma.+(s)
      
      
    }
   
   sigma
   
   
 } 
 
     def ic(a: Long): Double = {
       val d = neighbors.lookup(a).distinct.head.toSet
       if (d.isEmpty) { return 0.0 }
      else {
 
         return sigmaIc(d.toList)
       }
     }
 
    def mostICA(a: Long, b: Long): Double = {
 
       val an = neighbors.lookup(a).distinct.head.toSet
       val an1 = neighbors.lookup(b).distinct.head.toSet
       if (an.isEmpty || an1.isEmpty) { return 0.0 }
       val commonNeighbor = an.intersect(an1).toArray
      
      if (commonNeighbor.isEmpty) { return 0.0 }
      else {
        return sigmaIc(commonNeighbor.toList)
      }
 
     }
  
    
    /*
			 * Lin similarity measure 
			 */ 
     def simLin(e: Long, d: Long): Double = {
      if (ic(e) > 0.0 || ic(d) > 0.0) {
         (2.0.abs * (mostICA(e, d)).abs) / (ic(e).abs + ic(d).abs)
       } else { return 0.0 }
     }

    /*
			 * Jaccard similarity measure 
			 */
    def simJaccard(a: Long, b: Long): Double = {
      intersection(a, b) / union(a, b).toDouble

    }<commit_msg>Update RDFGraphPICClustering.scala<commit_after>package net.sansa_stack.ml.spark.clustering

import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.clustering.{ PowerIterationClusteringModel, PowerIterationClustering }
import org.apache.spark.graphx.{ Graph, EdgeDirection }
import scala.math.BigDecimal
import org.apache.commons.math3.util.MathUtils
import org.apache.spark.sql.SparkSession

class RDFGraphPICClustering(@transient val sparkSession: SparkSession,
                            val graph: Graph[Int, Int],
                            private val k: Int,
                            private val maxIterations: Int) extends Serializable {

  def clusterRdd(): RDD[(Long, Long, Double)] = {
    SimilaritesInPIC
  }

  /*
	 * Computes different similarities function for a given graph @graph.
	 */
  def SimilaritesInPIC(): RDD[(Long, Long, Double)] = {
    /*
	 * Collect all the edges of the graph
	*/
    val edge = graph.edges.collect()
    /*
	 * Collect neighbor IDs of all the vertices
	 */
    val neighbors = graph.collectNeighborIds(EdgeDirection.Either)
    /*
	 * Collect distinct vertices of the graph
	 */
    val vertices = graph.vertices.distinct()
    /*
	 * Difference between two set of vertices, used in different similarity measures
	 */
    def difference(a: Long, b: Long): Double = {
      val ansec = neighbors.lookup(a).distinct.head.toSet
      val ansec1 = neighbors.lookup(b).distinct.head.toSet
      if (ansec.isEmpty) { return 0.0 }
      val differ = ansec.diff(ansec1)
      if (differ.isEmpty) { return 0.0 }

      differ.size.toDouble
    }

    /*
	 * Intersection of two set of vertices, used in different similarity measures
	 */
    def intersection(a: Long, b: Long): Double = {
      val inters = neighbors.lookup(a).distinct.head.toSet
      val inters1 = neighbors.lookup(b).distinct.head.toSet
      if (inters.isEmpty || inters1.isEmpty) { return 0.0 }
      val rst = inters.intersect(inters1).toArray
      if (rst.isEmpty) { return 0.0 }
      rst.size.toDouble
    }

    /*
			 * Union of two set of vertices, used in different similarity measures
			 */
    def union(a: Long, b: Long): Double = {
      val inters = neighbors.lookup(a).distinct.head.toSet
      val inters1 = neighbors.lookup(b).distinct.head.toSet
      val rst = inters.union(inters1).toArray
      if (rst.isEmpty) { return 0.0 }

      rst.size.toDouble
    }
    // Logarithm base 2 
    val LOG2 = math.log(2)
    val log2 = { x: Double => math.log(x) / LOG2 }
	  
	  
	  val vertexCount = vertices.count()
	  
	  
	  // Similarity based on Information Theory
	  
	  
     var icc = 0.0
    val logC = MathUtils.log(10.0, vertexCount.toDouble)
 
     def informationContent(a: Long): Double = {
      if (a == 0) { return 0.0 }
       1 - (MathUtils.log(10.0, a.toDouble) / logC)
 
     }
     
	  def sigmaIc(a : List[Long]) : Double ={
   var sigma = 0.0
    for(k <- 0 until a.length) yield{
      
      val s = informationContent(a(k))
      sigma = sigma.+(s)
      
      
    }
   
   sigma
   
   
 } 
 
     def ic(a: Long): Double = {
       val d = neighbors.lookup(a).distinct.head.toSet
       if (d.isEmpty) { return 0.0 }
      else {
 
         return sigmaIc(d.toList)
       }
     }
     
	  
	  
	  def sigmaMICA(a : List[Long]) : Double ={
   var sigma = 0.0
    for(k <- 0 until a.length) yield{
      
      val s = ic(a(k))
      sigma = sigma.+(s)
      
      
    }
   
   sigma
   
   
 } 
    def mostICA(a: Long, b: Long): Double = {
 
       val an = neighbors.lookup(a).distinct.head.toSet
       val an1 = neighbors.lookup(b).distinct.head.toSet
       if (an.isEmpty || an1.isEmpty) { return 0.0 }
       val commonNeighbor = an.intersect(an1).toArray
      
      if (commonNeighbor.isEmpty) { return 0.0 }
      else {
        return sigmaMICA(commonNeighbor.toList)
      }
 
     }
  
    
    /*
			 * Lin similarity measure 
			 */ 
     def simLin(e: Long, d: Long): Double = {
      if (ic(e) > 0.0 || ic(d) > 0.0) {
         (2.0.abs * (mostICA(e, d)).abs) / (ic(e).abs + ic(d).abs)
       } else { return 0.0 }
     }

    /*
			 * Jaccard similarity measure 
			 */
    def simJaccard(a: Long, b: Long): Double = {
      intersection(a, b) / union(a, b).toDouble

    }
{'fraction_non_alphanumeric': 0.11964264155499216, 'fraction_numerical': 0.013642400096583364, 'mean_word_length': 2.1402577710386654, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '264329', 'n_tokens_mistral': 2931, 'n_tokens_neox': 2691, 'n_words': 663}