<commit_before>package net.sansa_stack.rdf.spark.model

import org.apache.spark.SparkContext
import org.apache.spark.graphx.PartitionStrategy.RandomVertexCut
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.{Graph => SparkGraph, EdgeTriplet, VertexId, Edge}

import scala.reflect.ClassTag
import scala.util.hashing.MurmurHash3

/**
 * Spark/GraphX based implementation of RDFGraphOps
 *
 * @author <NAME> <<EMAIL>>
 */
trait GraphXGraphOps[Rdf <: SparkGraphX{ type Blah = Rdf }]
  extends RDFGraphOps[Rdf] with Serializable { this: RDFNodeOps[Rdf] =>

  protected val sparkContext: SparkContext

  // graph
  def loadGraphFromNTriples(file: String, baseIRI: String): Rdf#Graph =
    makeGraph(sparkContext.textFile(file).mapPartitions {
      case it =>
        fromNTriples(it.mkString("\n"), baseIRI).iterator
    })

  def saveGraphToNTriples(graph: Rdf#Graph, file: String): Unit = {
    toTripleRDD(graph).mapPartitions {
      case it =>
        toNTriples(it.toIterable).split("\n").iterator
    }.saveAsTextFile(file)
  }

  def loadGraphFromSequenceFile(file: String): Rdf#Graph = loadGraphFromSequenceFile(file + ".vertices", file + ".edges")

  def saveGraphToSequenceFile(graph:Rdf#Graph, file: String): Unit = saveGraphToSequenceFile(graph, file + ".vertices", file + ".edges")

  // TODO: Do sequenceFile I/O using Avro, more efficient
  def loadGraphFromSequenceFile(vertexFile: String, edgeFile: String): Rdf#Graph = {
    val vertices: RDD[(VertexId, Rdf#Node)] = sparkContext.objectFile(vertexFile)
    val edges: RDD[Edge[Rdf#URI]] = sparkContext.objectFile(edgeFile)
    SparkGraph(vertices, edges)
  }

  // TODO: Do sequenceFile I/O using Avro, more efficient
  def saveGraphToSequenceFile(graph:Rdf#Graph, vertexFile: String, edgeFile: String): Unit = {
    graph.vertices.saveAsObjectFile(vertexFile)
    graph.edges.saveAsObjectFile(edgeFile)
  }

  def makeGraph(triples: Iterable[Rdf#Triple]): Rdf#Graph = {
    val triplesRDD = sparkContext.parallelize(triples.toSeq)
    makeGraph(triplesRDD)
  }

  def makeGraph(triples: RDD[Rdf#Triple]): Rdf#Graph = {
    val spo: RDD[(Rdf#Node, (Rdf#URI, Rdf#Node))] = triples.map {
      case Triple(s, p, o) => (s, (p, o))
    }

    val vertexIDs = spo.flatMap {
      case (s: Rdf#Node, (p: Rdf#URI, o: Rdf#Node)) =>
        Seq(s, p.asInstanceOf[Rdf#Node], o)
    }.zipWithUniqueId()

    val vertices: RDD[(VertexId, Rdf#Node)] = vertexIDs.map(v => (v._2, v._1))

    val subjectMappedEdges = spo.join(vertexIDs).map {
      case (s: Rdf#Node, ((p: Rdf#URI, o: Rdf#Node), sid: Long)) =>
        (o, (sid, p))
    }

    val subjectObjectMappedEdges: RDD[Edge[Rdf#URI]] = subjectMappedEdges.join(vertexIDs).map {
      case (o: Rdf#Node, ((sid: Long, p: Rdf#URI), oid: Long)) =>
        Edge(sid, oid, p)
    }

    val subjectVertexIds: RDD[(VertexId, Option[Int])] = subjectObjectMappedEdges.map(x => (x.srcId, None))
    val objectVertexIds: RDD[(VertexId, Option[Int])] = subjectObjectMappedEdges.map(x => (x.dstId, None))

    val a = SparkGraph(vertices, subjectObjectMappedEdges)
    a
  }

  def makeHashedVertexGraph(triples: RDD[Rdf#Triple]): Rdf#Graph = {
    val spo: RDD[(Rdf#Node, Rdf#URI, Rdf#Node)] = triples.map {
      case Triple(s, p, o) => (s, p, o)
    }

    def hash(s: Rdf#Node) = MurmurHash3.stringHash(s.toString).toLong
<commit_msg>Fix problems in GraphX building code for issue #4<commit_after>package net.sansa_stack.rdf.spark.model

import org.apache.spark.SparkContext
import org.apache.spark.graphx.PartitionStrategy.RandomVertexCut
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.{Graph => SparkGraph, EdgeTriplet, VertexId, Edge}

import scala.reflect.ClassTag
import scala.util.hashing.MurmurHash3

/**
 * Spark/GraphX based implementation of RDFGraphOps
 *
 * @author <NAME> <<EMAIL>>
 */
trait GraphXGraphOps[Rdf <: SparkGraphX{ type Blah = Rdf }]
  extends RDFGraphOps[Rdf] with Serializable { this: RDFNodeOps[Rdf] =>

  protected val sparkContext: SparkContext

  // graph
  def loadGraphFromNTriples(file: String, baseIRI: String): Rdf#Graph =
    makeGraph(sparkContext.textFile(file).mapPartitions {
      case it =>
        fromNTriples(it.mkString("\n"), baseIRI).iterator
    })

  def saveGraphToNTriples(graph: Rdf#Graph, file: String): Unit = {
    toTripleRDD(graph).mapPartitions {
      case it =>
        toNTriples(it.toIterable).split("\n").iterator
    }.saveAsTextFile(file)
  }

  def loadGraphFromSequenceFile(file: String): Rdf#Graph = loadGraphFromSequenceFile(file + ".vertices", file + ".edges")

  def saveGraphToSequenceFile(graph:Rdf#Graph, file: String): Unit = saveGraphToSequenceFile(graph, file + ".vertices", file + ".edges")

  // TODO: Do sequenceFile I/O using Avro, more efficient
  def loadGraphFromSequenceFile(vertexFile: String, edgeFile: String): Rdf#Graph = {
    val vertices: RDD[(VertexId, Rdf#Node)] = sparkContext.objectFile(vertexFile)
    val edges: RDD[Edge[Rdf#URI]] = sparkContext.objectFile(edgeFile)
    SparkGraph(vertices, edges)
  }

  // TODO: Do sequenceFile I/O using Avro, more efficient
  def saveGraphToSequenceFile(graph:Rdf#Graph, vertexFile: String, edgeFile: String): Unit = {
    graph.vertices.saveAsObjectFile(vertexFile)
    graph.edges.saveAsObjectFile(edgeFile)
  }

  def makeGraph(triples: Iterable[Rdf#Triple]): Rdf#Graph = {
    val triplesRDD = sparkContext.parallelize(triples.toSeq)
    makeGraph(triplesRDD)
  }

  def makeGraph(triples: RDD[Rdf#Triple]): Rdf#Graph = {
    val spo: RDD[(Rdf#Node, (Rdf#URI, Rdf#Node))] = triples.map(fromTriple).map {
      case (s, p, o) => (s, (p, o))
    }

    val vertexIDs = spo.flatMap {
      case (s, (p, o)) =>
        Seq(s, p.asInstanceOf[Rdf#Node], o)
    }.zipWithUniqueId()

    val vertices: RDD[(VertexId, Rdf#Node)] = vertexIDs.map(v => (v._2, v._1))

    val subjectMappedEdges = spo.join(vertexIDs).map {
      case (s, ((p, o), sid)) =>
        (o, (sid, p))
    }

    val subjectObjectMappedEdges: RDD[Edge[Rdf#URI]] = subjectMappedEdges.join(vertexIDs).map {
      case (o, ((sid, p), oid)) =>
        Edge(sid, oid, p)
    }

    val subjectVertexIds: RDD[(VertexId, Option[Int])] = subjectObjectMappedEdges.map(x => (x.srcId, None))
    val objectVertexIds: RDD[(VertexId, Option[Int])] = subjectObjectMappedEdges.map(x => (x.dstId, None))

    val a = SparkGraph(vertices, subjectObjectMappedEdges)
    a
  }

  def makeHashedVertexGraph(triples: RDD[Rdf#Triple]): Rdf#Graph = {
    val spo: RDD[(Rdf#Node, Rdf#URI, Rdf#Node)] = triples.map {
      case Triple(s, p, o) => (s, p, o)
    }

    def hash(s: Rdf#Node) = MurmurHash3.stringHash(s.toString).toLong

{'fraction_non_alphanumeric': 0.14255607406292337, 'fraction_numerical': 0.0013548095739876562, 'mean_word_length': 4.1543832428238945, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 33, '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': '836694', 'n_tokens_mistral': 2427, 'n_tokens_neox': 2355, 'n_words': 481}