go

distributed_social_network_analyzer/README.md

@@ -0,0 +1,44 @@
+# Distributed Social Network Analyzer #
+
+### Requirements ###
+
+Required sbt version: 1.4.6
+- Java 11
+- Spark 3.0.1 (with GraphX) : https://spark.apache.org/releases/spark-release-3-0-1.html
+- sbt project uses scala 2.12.2
+
+You can get help [here](http://www.scala-sbt.org/download.html)
+
+
+
+### Compiling ###
+
+Go to the project folder and execute
+```
+$sbt package
+```
+
+### Executing ###
+
+The main way to execute the project in Apache Spark after compiling:
+
+
+```
+$ spark-submit --master local[*] --class "MainAppFromXML" target/scala-2.12/social-network-analyzer_2.12-1.0.jar  xmlfilewithconversationgraph
+```
+
+
+Or use the script found in scripts/solveXMLconversation.sh
+
+
+# GraphX Pregel API
+
+## Doc:
+
+https://spark.apache.org/docs/latest/graphx-programming-guide.html#pregel-api
+
+## Examples:
+
+ spark-2.2.1-bin-hadoop2.7/examples/src/main/scala/org/apache/spark/examples/graphx/
+
+

distributed_social_network_analyzer/distributed-analyzer.sbt

@@ -0,0 +1,10 @@
+name := "Social Network Analyzer"
+
+version := "1.0"
+
+scalaVersion := "2.12.2"
+
+libraryDependencies += "org.apache.spark" %% "spark-core" % "3.0.1"
+libraryDependencies += "org.apache.spark" %% "spark-graphx" % "3.0.1"
+libraryDependencies += "org.json4s" %% "json4s-jackson" % "3.2.11"
+libraryDependencies += "org.apache.spark" %% "spark-mllib" % "3.0.1"

distributed_social_network_analyzer/scripts/solveXMLconversation.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+# Pass name of XML file with discussion graph, without .xml suffix
+spark-submit --master local[4] --class "MainAppFromXML" target/scala-2.12/social-network-analyzer_2.12-1.0.jar $1

distributed_social_network_analyzer/src/main/scala/DownloaderManagers/RedditDownloader.scala

@@ -0,0 +1,17 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+class RedditDownloader (context: SparkContext){
+  def getGraphFromConversationId(id:String, json_file:String) : RedditGraph = {
+    val resultGraph : RedditGraph = new RedditGraph(context)
+    val scriptPath = "Downloaders/reddit-collector/reddit.py"
+    val scriptName = "reddit.py"
+    context.addFile(scriptPath)
+    val data = context.parallelize(List(id))
+    val pipeRDD = data.pipe(scriptPath)
+    pipeRDD.collect().foreach(println)
+    resultGraph.parseRedditConversationFromFile(json_file)
+    resultGraph
+  }
+}

distributed_social_network_analyzer/src/main/scala/DownloaderManagers/TwitterDownloader.scala

@@ -0,0 +1,27 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+class TwitterDownloader (context: SparkContext){
+  // def getGraphFromConversationId(id:String) : TwitterGraph = {
+  //   val resultGraph : TwitterGraph = new TwitterGraph(context)
+  //   val scriptPath = "Downloaders/twitter-collector/twcollect.py"
+  //   val scriptName = "twcollect.py"
+  //   context.addFile(scriptPath)
+  //   val data = context.parallelize(List(id))
+  //   val pipeRDD = data.pipe(scriptPath)
+  //   pipeRDD.collect()
+  //   resultGraph.parseConversationFromFile("Storage/Jsons/" + id + ".json")
+  //   resultGraph
+  // }
+
+  def downloadAndStoreConversation(id: String) : String = {
+    val scriptPath = "Downloaders/twitter-collector/twcollect.py"
+    val scriptName = "twcollect.py"
+    context.addFile(scriptPath)
+    val data = context.parallelize(List(id))
+    val pipeRDD = data.pipe(scriptPath)
+    pipeRDD.collect()
+    "Storage/Jsons/" + id + ".json"
+  }
+}

distributed_social_network_analyzer/src/main/scala/Graph/RedditGraph.scala

@@ -0,0 +1,59 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//Json4s
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+/***Summary***
+RedditGraph is the specific parser for reddit, it takes a Reddit
+conversation or a specific reddit jsons (obtained directly from reddit API)
+and SetUp a SocialGraph with this information.
+*************/
+
+class RedditGraph(context: SparkContext) extends SocialGraph(context){
+
+  def parseRedditConversationFromFile(filePath: String){
+    val file = scala.io.Source.fromFile(filePath)
+    val lines = file.getLines()
+    for (line <- lines){
+      parseRedditMessage(line)
+    }
+    file.close()
+  }
+
+  def parseRedditConversation(jsonConversation: String){
+    val conversation = parse(jsonConversation).extract[Reddit_Conversation_Messages]
+    val conversation_properties = parse(jsonConversation).extract[Reddit_Conversation_Properties]
+    for(i <- 0 until conversation.messages.size){
+      val l_message = conversation.messages(i)
+      val l_properties = conversation_properties.messages(i)
+      println("Message readed:"+
+                            "\n\tId:"       + l_message.id      +
+                            "\n\tAuthor: "  + l_message.author_name  +
+                            "\n\tMessage: " + l_message.text)
+      addMessage(l_message.id, l_message.author_name, l_message.text, l_properties)
+      if(l_message.parent != -1){
+        addConnection(l_message.id, l_message.parent, "noone") //TODO: Add real intention
+      }
+    }
+  }
+
+  def parseRedditMessage(jsonMessage: String){
+    parseRedditConversation("{\"messages\":["+jsonMessage+"]}")
+  }
+
+}
+
+case class Reddit_Temp(id: Long, author_name: String, text: String, parent : Long)
+case class Reddit_Conversation_Messages(messages: List[Reddit_Temp])
+case class Reddit_Conversation_Properties(messages: List[Map[String, Any]])

distributed_social_network_analyzer/src/main/scala/Graph/SocialGraph.scala

@@ -0,0 +1,364 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Json4s
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+import org.json4s.jackson.Serialization
+
+//Lists
+import scala.collection.mutable.ListBuffer
+// import scala.collection.mutable
+
+//Serialization
+import java.io._
+
+
+
+/***Summary***
+JsonGraph it's and abstraction for Graph creation, allowing creating messages
+and connections that could be added directly or parsed from a Json
+
+Every message must contain an identifier an author, a message and a Json filled
+with properties.
+
+Every connection will represent the edge between two messages, represented by
+source and destination identifiers and the kind of relation between them.
+************/
+
+class SocialGraph(context: SparkContext){
+    var m_context = context
+    var m_messages : Map[Long, Message] = Map() //ID -> Message
+    var m_connections : Map[(Long, Long), Edge[String]] = Map() //(From, To) -> Edge
+    var g_graph : Option[Graph[Message, String]] = None
+    var subgraph_indegrees = scala.collection.mutable.Map[Long, Int]() //   Map IDvertex to its INdegree in subgraph
+    var subgraph_defindegrees = scala.collection.mutable.Map[Long, Int]() //   Map IDvertex to its INdegree in subgraph
+    val m_representations = new GraphRepresentations(this, context)
+    var listofdiscnodes =  new ListBuffer[String]()
+
+    implicit val formats = DefaultFormats
+
+    def getSubGraph(attack: Boolean , support: Boolean, noone: Boolean) : SocialGraph = {
+      var resultGraph : SocialGraph = new SocialGraph(context)
+      var subGraphMessages : Map[Long, Message] = Map()
+      var sweight : Float = 0;
+      var dweight : Float = 0;
+
+      listofdiscnodes =  new ListBuffer[String]()
+      // Initialize degrees
+      resultGraph.subgraph_indegrees = scala.collection.mutable.Map[Long, Int]()
+      resultGraph.subgraph_defindegrees = scala.collection.mutable.Map[Long, Int]()
+      for (msg <- m_messages.values.toList) {
+        resultGraph.subgraph_indegrees += (msg.id -> 0)
+        resultGraph.subgraph_defindegrees += (msg.id -> 0)
+      }
+
+       //Look for desired connections
+       for (connection <- m_connections.values.toList){
+        if((attack && connection.attr == "attack") ||
+          (support && connection.attr == "support") ||
+          (noone && connection.attr == "noone")){
+            println( "Connection :"+connection.srcId.toString()+ " -> "+connection.dstId.toString())
+            val srcMessage : Message = m_messages(connection.srcId)
+            val dstMessage : Message = m_messages(connection.dstId)
+            sweight = srcMessage.weight
+            dweight = dstMessage.weight
+            if(!subGraphMessages.contains(connection.srcId)){
+
+              subGraphMessages += (connection.srcId -> srcMessage)
+              resultGraph.addMessage(srcMessage.id, srcMessage.author, srcMessage.message,
+              srcMessage.properties, srcMessage.weight)
+              // println( "Adding "+srcMessage.id.toString() )
+
+            }
+            if(!subGraphMessages.contains(connection.dstId)){
+
+              subGraphMessages += (connection.dstId -> dstMessage)
+              resultGraph.addMessage(dstMessage.id, dstMessage.author, dstMessage.message,
+              dstMessage.properties, dstMessage.weight)
+              // println( "Adding "+dstMessage.id.toString() )
+
+            }
+            resultGraph.subgraph_indegrees(connection.dstId) = resultGraph.subgraph_indegrees(connection.dstId) + 1
+            if (sweight.>=( dweight )) {
+              resultGraph.subgraph_defindegrees(connection.dstId) = resultGraph.subgraph_defindegrees(connection.dstId) + 1
+               println( "Effective : "+connection.srcId+"-"+sweight+ " -> "+connection.dstId+"-"+dweight)
+            } else {
+                 println( "NOT Effective : "+connection.srcId+"-"+sweight+ " -> "+connection.dstId+"-"+dweight)
+              }
+            resultGraph.addConnection(connection.srcId, connection.dstId, connection.attr)
+        }
+      }
+      var ndis : Int = 0;
+      // subGraphMessages.foreach(println)
+      for (msg <- m_messages.values.toList) {
+         if (!subGraphMessages.contains(msg.id)) {
+           val idstr = msg.id.toString()
+           // println(  " Node desconnected from graph: " + idstr  )
+           listofdiscnodes += idstr
+           ndis = ndis + 1;
+        }
+      }
+      //for (msg <- m_messages.values.toList) {
+
+        //   println(  " Node  "+ msg.id.toString() +" indegre: " + resultGraph.subgraph_indegrees(msg.id).toString())
+
+      // }
+      println( "Total disconnected : "+ndis.toString() )
+      resultGraph
+    }
+
+    def getListOfDisconnectedNodes() : ListBuffer[String] = {
+      listofdiscnodes
+    }
+
+    def updatePunctuation(id : Long, newPunctuation : Float){
+      m_messages(id).weight = newPunctuation
+    }
+
+    def storeGraph(directory : String, graphName : String){
+      //Create files
+      val file_messages = new File(directory + graphName + ".mes")
+      val file_connections = new File(directory + graphName + ".con")
+      val writer_messages = new BufferedWriter(new FileWriter(file_messages))
+      val writer_connections = new BufferedWriter(new FileWriter(file_connections))
+      var exp_json : String = ""
+      var map_json : String = ""
+
+      for (message <- m_messages.values.toList){
+
+          exp_json += "{\"id\":"        + message.id              + ","   +
+                      "\"weight\":"     + message.weight          + ","   +
+                      "\"author\":\""   + message.author          + "\"," +
+                      "\"message\":\""  + message.message.replace("\"", "\\\"").replace("\n", " ") + "\"," +
+                      "\"properties\":" + Serialization.write(message.properties)  + "}"
+          writer_messages.write(exp_json)
+          exp_json = "\n"
+      }
+
+      exp_json = ""
+      for (connection <- m_connections.values.toList){
+        exp_json += "{\"origin\":"     + connection.srcId     + "," +
+                   "\"destination\":" + connection.dstId      + "," +
+                   "\"intention\":\"" + connection.attr       + "\"}"
+        writer_connections.write(exp_json)
+        exp_json = "\n"
+      }
+
+      //Close files
+      writer_messages.close()
+      writer_connections.close()
+    }
+
+    def loadGraph(directory : String, graphName : String){
+      //Open files
+      val baseFilename = directory + graphName
+      val file_messages = scala.io.Source.fromFile(baseFilename + ".mes")
+      val file_connections = scala.io.Source.fromFile(baseFilename + ".con")
+
+      //Load messages
+      var lines = file_messages.getLines()
+      var message : String = ""
+      var count : Int = 0
+
+      for (line <- lines){
+          parseMessage(line)
+          count += 1
+      }
+
+      //Load connections
+      lines = file_connections.getLines()
+      for (line <- lines){
+        parseConnection(line)
+      }
+
+      //Close files
+      file_messages.close()
+      file_connections.close()
+
+    }
+
+    def makeGraph(){
+      //Create vertex
+      val g_vertex : RDD[(VertexId, Message)] = m_context.parallelize(messagesVertexFormat())
+      //Create edges
+      val g_edges : RDD[Edge[String]] = m_context.parallelize(m_connections.values.to[collection.immutable.Seq])
+      //Create default if no relationship
+      val defaultMap : Map[String, Any] = Map("Invalid" -> "Invalid")
+      val defaultRelation = new Message(-1L, 0, "Invalid", "Invalid", defaultMap)
+
+      //Build the graph
+      g_graph = Some(Graph(g_vertex, g_edges, defaultRelation))
+    }
+
+    def getGraph(): Graph[Message, String] = {
+      g_graph match{
+        case None => {
+          makeGraph()
+          getGraph()
+        }
+        case Some(s) => s
+      }
+    }
+
+    def addMessage(id: Long, author: String, message: String,
+       properties: Map[String, Any], weight : Float = -1){
+      m_messages += (id -> new Message(id, weight, author, message, properties))
+    }
+
+    def addConnection(origin: Long, destination: Long, intention: String){
+      m_connections += ((origin, destination) -> new Edge(origin, destination, intention))
+    }
+
+    /***Summary***
+    Calculates the max of the distances of the graph.
+    *************/
+    def maxDistance() : Int = {
+      val graph = m_representations.integerOnly(0)
+      val sssp = graph.pregel(0)(
+        (id, dist, newDist) => {
+          newDist
+        },
+        triplet => {
+          if(triplet.dstAttr <= triplet.srcAttr){
+            Iterator((triplet.dstId, triplet.srcAttr+1))
+          }
+          else{
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          if(a>b) a
+          else b
+        }
+      )
+      val distance = sssp.vertices.reduce((a,b) => {
+        if(a._2 > b._2) a
+        else b
+      })
+      println(sssp.vertices.collect.mkString("\n"))
+      distance._2
+    }
+
+    /***Summary***
+    Calculates the min of the distances of the graph.
+    *************/
+    def minDistance() : Int = {
+      val graph = m_representations.integerBool(0, true)
+      val sssp = graph.pregel((0, true))(
+        (id, dist, newDist) => {
+          if(dist._1 == newDist._1 && dist._1 != 0){
+            (dist._1, false)
+          }
+          else{
+            newDist
+          }
+        },
+        triplet => {
+          if(triplet.dstAttr._2){
+            Iterator((triplet.dstId, (triplet.srcAttr._1+1, true)))
+          }
+          else{
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          if(a._1 < b._1) a
+          else b
+        }
+      )
+      val distance = sssp.vertices.reduce((a,b) => {
+        if(a._2._1 > b._2._1) a
+        else b
+      })
+      println(sssp.vertices.collect.mkString("\n"))
+      distance._2._1
+    }
+
+    /***Summary***
+    Calculates the average of the distances of the graph.
+    *************/
+    def averageDistance() : Float = {
+
+      //First acquire the summ of all the distances
+      val graph = m_representations.integerIntegerBool()
+      val sssp = graph.pregel((-1, -1, true))(
+        (id, dist, newDist) => {
+          if(newDist._1 == -1){
+            (0,0,true)
+          }
+          else{
+            if(dist._1 == newDist._1 && dist._2 == newDist._2){
+              (dist._1, dist._2, false)
+            }
+            else{
+              newDist
+            }
+          }
+        },
+        triplet => {
+          if(triplet.dstAttr._3){
+            if(triplet.srcAttr._1 == 0){
+              Iterator((triplet.dstId, (1, 1, true)))
+            }
+            else{
+              Iterator((triplet.dstId, (triplet.srcAttr._1+triplet.srcAttr._2, triplet.srcAttr._2, true)))
+            }
+          }
+          else{
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          (a._1+b._1, a._2+b._2, true)
+        }
+      )
+      val distance = sssp.vertices.reduce((a,b) => {
+        if(a._2._1 > b._2._1) a
+        else b
+      })
+      //println(sssp.vertices.collect.mkString("\n"))
+      //Divide total distances between all the leafs.
+      val leafs = (graph.vertices.count - graph.inDegrees.count)
+      //println("Total leafs: " + leafs)
+      //println("Total distance " + distance._2._1)
+      val avg : Float = (distance._2._1.toFloat/leafs.toFloat)
+      //Return average
+      avg
+    }
+
+    def getMessages(): Map[Long, Message] = m_messages
+
+    def getConnections(): Map[(Long, Long), Edge[String]] = m_connections
+
+    def getRepresentation(): GraphRepresentations = m_representations
+
+    def parseMessage(messageJson : String){
+      val parsedMessage : Message = parse(messageJson).extract[Message]
+      m_messages += (parsedMessage.id -> parsedMessage)
+    }
+
+    def parseConnection(connectionJson : String){
+      val conn = parse(connectionJson).extract[Connection]
+      m_connections += ((conn.origin, conn.destination) -> new Edge(conn.origin, conn.destination, conn.intention))
+    }
+
+    private def messagesVertexFormat() : Array[(VertexId, Message)] = {
+      val returnArray = new Array[(VertexId, Message)](m_messages.toList.length)
+      val messagesArray = m_messages.toList.toArray
+      for(i <- 0 until messagesArray.length){
+        returnArray(i) = (messagesArray(i)._1, messagesArray(i)._2)
+      }
+      returnArray
+    }
+
+}
+
+case class Message(id: Long, var weight: Float, author: String, message: String, properties: Map[String, Any])
+case class Connection (origin: Long, destination: Long, intention: String)

distributed_social_network_analyzer/src/main/scala/Graph/TwitterGraph.scala

@@ -0,0 +1,148 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//Json4s
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+import scala.xml.XML
+
+/***Summary***
+TwitterGraph is the specific parser for twitter, it takes a Twitter
+conversation or a specific tweets jsons (obtained directly from twitter API)
+and SetUp a SocialGraph with this information.
+*************/
+
+/***Conversation Json***
+There are two main ways to parse a conversation:
+
+1st: Do it tweet by tweet: place on every line a json from a tweet from the
+conversation and read it line by line and parse them using parseTweet.
+
+2nd: Using a conversation json that must be composed on that way:
+{"tweets":[{jsonTweet1}, {jsonTweet2}, {jsonTweet3}]}
+Where a jsonTweet is a json of a tweet obtained directly from twitter API
+***********************/
+
+class TwitterGraph(context: SparkContext, classifier: MPC) extends SocialGraph(context){
+  var userReferences : Map[String, List[Long]] = Map()
+  var focusTo : Map[Long, Option[Long]] = Map().withDefaultValue(None)
+  var tweetAuthor : Map[Long, String] = Map()
+
+
+
+  def parseConversationFromFile(filePath: String){
+    val file = context.textFile(filePath).toLocalIterator.toArray
+    for (line <- file){
+      parseTweet(line)
+    }
+  }
+
+  def parseConversation(jsonConversation: String){
+    val conversation = parse(jsonConversation).extract[Twitter_Conversation_Tweets]
+    val conversation_properties = parse(jsonConversation).extract[Twitter_Conversation_Properties]
+    for(i <- 0 until conversation.tweets.size){
+      val l_tweet = conversation.tweets(i)
+      val l_properties = conversation_properties.tweets(i)
+      println("Tweet readed:" +
+              "\n\tId:"       + l_tweet.id      +
+              "\n\tAuthor: "  + l_tweet.user("screen_name").toString  +
+              "\n\tMessage: " + l_tweet.text)
+      addMessage(l_tweet.id, l_tweet.user("screen_name").toString,
+      l_tweet.text, l_properties, weightCalc(l_tweet.user("followers_count").asInstanceOf[Number].intValue,
+      l_tweet.retweet_count.toInt, l_tweet.favorite_count.toInt))
+      tweetAuthor += (l_tweet.id -> l_tweet.user("screen_name").toString)
+      println("Author of this tweet is: " + tweetAuthor(l_tweet.id))
+      makeLinks(l_tweet.id, l_properties, l_tweet.entities.user_mentions)
+    }
+  }
+
+
+  private def weightCalc(followers : Int, retweets : Int, favorites : Int) : Float = {
+    var sum : Double = followers
+    sum += 20 * retweets
+    sum += 40 * favorites
+    sum += 1
+    (scala.math.log(sum)/scala.math.log(2)).toFloat
+  }
+
+  def parseTweet(jsonTweet: String){
+    parseConversation("{\"tweets\":["+jsonTweet+"]}")
+  }
+
+  private def makeLinks(id: Long, info: Map[String, Any],
+    mentions: List[Map[String,Any]]){
+    //Make basic link
+    val reply_to = info("in_reply_to_status_id")
+    if(reply_to != null){
+      val l_reply_to:Long = reply_to.asInstanceOf[Number].longValue
+      focusTo += (id -> Some(l_reply_to))
+      println("Link info:"+
+              "\n\tOrigin: " + id +
+              "\n\tDestin: " + l_reply_to)
+      addConnection(id, l_reply_to, classifier.getRelationship(id,l_reply_to))
+      val mentionList : ListBuffer[String] = getMentionList(mentions)
+      println("Mention list: ")
+      for(mention <- mentionList){
+        println("\t-" + mention)
+      }
+      makeFullLinks(id, l_reply_to, mentionList)
+    }
+    else{
+      println("This is the root tweet")
+    }
+  }
+
+  private def makeFullLinks(id: Long, reply_to: Long,
+    mentions: ListBuffer[String]){
+    var nextLookFocus = reply_to
+    var focus: Option[Long] = None
+    for(mention <- mentions){
+      do {
+       focus = focusTo.getOrElse(nextLookFocus, None)
+       focus match{
+         case Some(s) => {
+           val currAuthor = tweetAuthor.getOrElse(s, None)
+           if(currAuthor != None && currAuthor == mention && reply_to != s){
+             println(println("Link info:"+
+                     "\n\tOrigin: " + id +
+                     "\n\tDestin: " + s))
+           }
+           addConnection(id, s, classifier.getRelationship(id, s))
+           nextLookFocus = s
+         }
+         case None => {
+           nextLookFocus = reply_to
+         }
+       }
+      }
+      while(focus != None)
+    }
+  }
+
+  private def getMentionList(mentions: List[Map[String, Any]]) : ListBuffer[String]= {
+    val mentionList: ListBuffer[String] = ListBuffer[String]()
+    for(mention <- mentions){
+      val name : String = mention("screen_name").toString
+      if(!mentionList.contains(name))
+        mentionList += name
+    }
+    mentionList.distinct
+    mentionList
+  }
+
+}
+
+case class Tweet_Temp(id: Long, user: Map[String, Any], text: String,
+  entities : Entities_Temp, retweet_count : Long, favorite_count : Long)
+case class Entities_Temp(user_mentions: List[Map[String, Any]])
+case class Twitter_Conversation_Tweets(tweets: List[Tweet_Temp])
+case class Twitter_Conversation_Properties(tweets: List[Map[String, Any]])

distributed_social_network_analyzer/src/main/scala/Graph/TwitterGraphFromXML.scala

@@ -0,0 +1,78 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//Json4s
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+import scala.xml.XML
+
+
+
+/***Summary***
+TwitterGraph is the specific parser for twitter, it takes a Twitter
+conversation or a specific tweets jsons (obtained directly from twitter API)
+and SetUp a SocialGraph with this information.
+*************/
+
+/***Conversation Json***
+There are two main ways to parse a conversation:
+
+1st: Do it tweet by tweet: place on every line a json from a tweet from the
+conversation and read it line by line and parse them using parseTweet.
+
+2nd: Using a conversation json that must be composed on that way:
+{"tweets":[{jsonTweet1}, {jsonTweet2}, {jsonTweet3}]}
+Where a jsonTweet is a json of a tweet obtained directly from twitter API
+***********************/
+
+class TwitterGraphFromXML(context: SparkContext ) extends SocialGraph(context){
+  var userReferences : Map[String, List[Long]] = Map()
+  var focusTo : Map[Long, Option[Long]] = Map().withDefaultValue(None)
+  var tweetAuthor : Map[Long, String] = Map()
+
+
+
+
+  def getMeaningOfLabel(relation: String ): String = {
+    val meaning = relation match{
+      case "ATTACKS" => "attack"
+      case "SUPPORTS" => "support"
+      case _ => "noone"
+    }
+    return meaning
+  }
+
+
+  def parseXMLLabelledConversation( xmlLabelledConversation: String ) {
+     val xmlconv = XML.loadFile( xmlLabelledConversation )
+     // Parse the argument-list list of <arg > elements
+     val arguments = xmlconv \ "argument-list"
+     val listofarguments = arguments \ "arg"
+     for (arg <- listofarguments) {
+       //  println( (arg \ "@id").text + arg.text + (arg \ "@weight").text)
+       addMessage( (arg \ "@id").text.toLong, "anonymous",
+       arg.text, arg.attributes.asAttrMap, (arg \ "@weight").text.toFloat )
+     }
+     // Parse the argument pairs
+     val argpairs =  xmlconv \ "argument-pairs"
+     val listofargpairs = argpairs \ "pair"
+     for (apair <- listofargpairs) {
+       // println( apair )
+       val answer = ((apair \ "t") \"@id" ).text
+       val source = ((apair \ "h") \"@id" ).text
+       // println( answer + " -> " + source + " " + getMeaningOfLabel((apair \ "@entailment").text) )
+       addConnection(  answer.toLong, source.toLong, getMeaningOfLabel((apair \ "@entailment").text))
+     }
+  }
+
+
+}

distributed_social_network_analyzer/src/main/scala/GraphCalculous/GraphRepresentations.scala

@@ -0,0 +1,220 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+class GraphRepresentations(val originalGraph: SocialGraph,
+                           val context: SparkContext){
+
+    /***Summary***
+    This class provides different graph representations from SocialGraph
+    ************/
+
+    /**Summary**
+    Gives a Integer graph representation
+    with the same struct and connections of the original.
+    ***********/
+    def integerOnly(initValue : Integer): Graph[Int, String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, Int)] =
+        new ListBuffer[(VertexId, Int)]
+      originalGraph.getMessages().map(a => replicatedVertices += ((a._1, initValue)))
+      val g_vertex : RDD[(VertexId, Int)] = context.parallelize(replicatedVertices)
+      //Replicate connections
+      val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+      //Build the graph
+      Graph(g_vertex, g_edges)
+    }
+
+    /**Summary**
+    Gives a Integer,Integer graph representation
+    with the same struct and connections of the original.
+    ***********/
+    def integerInteger(): Graph[(Int, Int), String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, (Int, Int))] =
+        new ListBuffer[(VertexId, (Int, Int))]
+      originalGraph.getMessages().map(a => replicatedVertices += ((a._1, (0, 0))))
+      val g_vertex : RDD[(VertexId, (Int, Int))] = context.parallelize(replicatedVertices)
+      //Replicate connections
+      val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+      //Create default if no relationship
+      //val defaultRelation = 0
+      //Build the graph
+      Graph(g_vertex, g_edges)//, defaultRelation)
+    }
+
+    /**Summary**
+    Gives a Integer,Bool graph representation
+    with the same struct and connections of the original
+    in order to i.e. stop sending messages to a node marked as 'false'
+    in pregel.
+    ***********/
+    def integerBool(intVal:Integer, boolVal:Boolean): Graph[(Int, Boolean), String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, (Int, Boolean))] =
+        new ListBuffer[(VertexId, (Int, Boolean))]
+        originalGraph.getMessages().map(a => replicatedVertices += ((a._1, (intVal, boolVal))))
+        val g_vertex : RDD[(VertexId, (Int, Boolean))] = context.parallelize(replicatedVertices)
+        //Replicate connections
+        val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+        //Create default if no relationship
+        //val defaultRelation = 0
+        //Build the graph
+        Graph(g_vertex, g_edges)//, defaultRelation)
+    }
+
+    /**Summary**
+    Gives a Integer,Integer,Boolean graph representation
+    with the same struct and connections of the original.
+    ***********/
+    def integerIntegerBool(): Graph[(Int, Int, Boolean), String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, (Int, Int, Boolean))] =
+        new ListBuffer[(VertexId, (Int, Int, Boolean))]
+      originalGraph.getMessages().map(a => replicatedVertices += ((a._1, (0, 0, true))))
+      val g_vertex : RDD[(VertexId, (Int, Int, Boolean))] = context.parallelize(replicatedVertices)
+      //Replicate connections
+      val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+      //Create default if no relationship
+      //val defaultRelation = 0
+      //Build the graph
+      Graph(g_vertex, g_edges)//, defaultRelation)
+    }
+
+    /**Summary**
+    Gives a Float,Integer,Boolean graph representation
+    with the same struct and connections of the original.
+    The first value can be setted with the weight of the node.
+    ***********/
+    def FloatIntegerBool(first: Float = 0, second: Integer = 0,
+      third: Boolean = true, firstIsWeight: Boolean = false,
+      leafsBool: Boolean = true):
+      Graph[(Float, Int, Boolean), String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, (Float, Int, Boolean))] =
+        new ListBuffer[(VertexId, (Float, Int, Boolean))]
+      if(firstIsWeight){
+        originalGraph.getMessages().map(a => {
+          var initValue : Boolean = leafsBool
+          if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+            initValue = third
+          }
+          replicatedVertices += ((a._1, (a._2.weight, second, initValue)))
+        })
+      }
+      else{
+        originalGraph.getMessages().map(a => {
+          var initValue : Boolean = leafsBool
+          if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+            initValue = third
+          }
+          replicatedVertices += ((a._1, (first, second, initValue)))
+        })
+      }
+      val g_vertex : RDD[(VertexId, (Float, Int, Boolean))] = context.parallelize(replicatedVertices)
+      //Replicate connections
+      val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+      //Create default if no relationship
+      //val defaultRelation = 0
+      //Build the graph
+      Graph(g_vertex, g_edges)//, defaultRelation)
+    }
+
+      def FloatIntegerLongBool( first: Float = 0,
+                                second: Integer = 0,
+                                third: Long = 0,
+                                fourth: Boolean = true,
+                                firstIsWeight: Boolean = false,
+                                leafsBool: Boolean = true):
+      Graph[(Float, Int, Long, Boolean), String] = {
+      //Replicate vertices
+      var replicatedVertices : ListBuffer[(VertexId, (Float, Int, Long, Boolean))] =
+        new ListBuffer[(VertexId, (Float, Int, Long, Boolean))]
+      if(firstIsWeight){
+        originalGraph.getMessages().map(a => {
+          var initValue : Boolean = leafsBool
+          if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+            initValue = fourth
+          }
+          replicatedVertices += ((a._1, (a._2.weight, second, third, initValue)))
+        })
+      }
+      else{
+        originalGraph.getMessages().map(a => {
+          var initValue : Boolean = leafsBool
+          if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+            initValue = fourth
+          }
+          replicatedVertices += ((a._1, (first, second, third, initValue)))
+        })
+      }
+      val g_vertex : RDD[(VertexId, (Float, Int, Long, Boolean))] = context.parallelize(replicatedVertices)
+      //Replicate connections
+      val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+      //Create default if no relationship
+      //val defaultRelation = 0
+      //Build the graph
+      Graph(g_vertex, g_edges)//, defaultRelation)
+    }
+
+
+    // Compute vertex representation as:
+    //
+    //  ((Weight,INdegree,DEFsindegree), (NumAcceptedAttackers,NumRejectedAttackers), NumReceivedMessages, Finished)
+    //
+    def FloatIntegerIntegerIntegerLongBool(
+        firstIsWeight: Boolean = false,
+        first: Float = 0,
+        second: Integer = 0, third: Integer = 0,  // (NumAcceptedAttackers,NumRejectedAttackers)
+        fourth: Boolean = false,                   // Finished default value
+        leafsInitState: Boolean = true  ) :
+    Graph[( (Float,Int,Int), (Int,Int), Long, Boolean), String] = {
+    //Replicate vertices
+    var replicatedVertices : ListBuffer[(VertexId, ((Float,Int,Int), (Int,Int), Long, Boolean))] =
+      new ListBuffer[(VertexId, ((Float,Int,Int), (Int,Int), Long, Boolean))]
+
+      //println( "INDEGREES: "+originalGraph.subgraph_indegrees.values.toList.toString())
+      //for (msg <- originalGraph.m_messages.values.toList) {
+
+        //   println(  " Node  "+ msg.id.toString() +" indegre: " + originalGraph.subgraph_indegrees(msg.id).toString())
+
+      // }
+
+    if(firstIsWeight){
+      originalGraph.getMessages().map(a => {
+        // println( " ID of vertex processed: "+a._1.toString())
+        var initValue : Boolean = leafsInitState
+        if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+          initValue = fourth
+        }
+        replicatedVertices += ((a._1, ((a._2.weight,originalGraph.subgraph_indegrees(a._1),originalGraph.subgraph_defindegrees(a._1)), (second,third), 0, initValue)))
+        // println( "Vertex processed")
+      })
+    }
+    else{
+      originalGraph.getMessages().map(a => {
+        var initValue : Boolean = leafsInitState
+        if(originalGraph.getGraph().inDegrees.lookup(a._1) != Seq.empty[Int]){
+          initValue = fourth
+        }
+        replicatedVertices += ((a._1, ((first,originalGraph.subgraph_indegrees(a._1),originalGraph.subgraph_defindegrees(a._1)), (second,third), 0, initValue)))
+      })
+    }
+    val g_vertex : RDD[(VertexId, ((Float,Int,Int), (Int,Int), Long, Boolean))] = context.parallelize(replicatedVertices)
+    //Replicate connections
+    val g_edges : RDD[Edge[String]] = context.parallelize(originalGraph.getConnections().values.to[collection.immutable.Seq])
+    //Create default if no relationship
+    //val defaultRelation = 0
+    //Build the graph
+    Graph(g_vertex, g_edges)//, defaultRelation)
+  }
+
+}

distributed_social_network_analyzer/src/main/scala/GraphCalculous/RelationshipsCalcs.scala

@@ -0,0 +1,264 @@
+//SparkContext
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+class RelationshipsCalcs(){
+
+    /*****Summary*****
+    |Graph represents Weight / Defeaters / Finished
+    |Defeaters: -1 at initial state, to set it's value at least one time,
+    |avoiding mark them as finished
+    |Finished: True / False
+    *****************/
+    def calcDefeaters(sGraph: SocialGraph) : Int = {
+      // Init graph to (Weight, -1, False)
+      val graph = sGraph.getSubGraph(true, false, false).getRepresentation().FloatIntegerBool(0, -1, false, true, true)
+
+      val sssp = graph.pregel((0, 0, false))(
+        (id, main, received) => {
+          //Evaluate the joined messages and set the new state
+          if((main._2 == received._2 && received._3) || main._3){
+            //If the state it's the same as previous mark as finished
+            (main._1, received._2, true)
+          }
+          else{
+            (main._1, received._2, false)
+          }
+        },
+        triplet => {
+          //Compute the messages to send
+          if(triplet.srcAttr._1 >= triplet.dstAttr._1 && //Weight
+             triplet.dstAttr._3 == false){               //Dst is not finished
+            //Send a message
+            if(triplet.srcAttr._2 == 0){
+              Iterator((triplet.dstId, (0, 1, triplet.srcAttr._3)))
+            }
+            else{
+              Iterator((triplet.dstId, (0, 0, triplet.srcAttr._3)))
+            }
+          }
+          else{
+            //Don't send a message
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          //Join messages into one
+          (0, a._2 + b._2, a._3 && b._3)
+        }
+      )
+      val defeaterNodes = sssp.vertices.map(node => {
+        if(node._2._2 > 0){
+          0
+        }
+        else{
+          1
+        }
+      }).reduce((a, b) => a+b)
+      println("TOTAL DEFEATERS: " + defeaterNodes)
+      defeaterNodes
+
+    }
+
+
+    def optimizedCalcDefeaters(sGraph: SocialGraph) : Int = {
+      // Init graph to (Weight, -1, False)
+      val graph = sGraph.getSubGraph(true, false, false).getRepresentation().FloatIntegerBool(0, -1, false, true, true)
+
+      val sssp = graph.pregel((0, 0, false))(
+        (id, main, received) => {
+          //Evaluate the joined messages and set the new state
+          if((main._2 == received._2) || main._3){
+            //If the state it's the same as previous mark as finished
+            (main._1, received._2, true)
+          }
+          else{
+            if(!received._3){
+              (main._1, received._2, false)
+            }
+            else{
+              (main._1, received._2, true)
+            }
+          }
+        },
+        triplet => {
+          //Compute the messages to send
+          if(triplet.srcAttr._1 >= triplet.dstAttr._1 && //Weight
+             triplet.srcAttr._3 && !triplet.dstAttr._3){               //Dst is not finished
+            //Send a message
+            if(triplet.srcAttr._2 == 0){
+              Iterator((triplet.dstId, (0, 1, triplet.srcAttr._3)))
+            }
+            else{
+              Iterator((triplet.dstId, (0, 0, triplet.srcAttr._3)))
+            }
+          }
+          else{
+            //Don't send a message
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          //Join messages into one
+          (0, a._2 + b._2, true)
+        }
+      )
+      val defeaterNodes = sssp.vertices.map(node => {
+        if(node._2._2 > 0){
+          0
+        }
+        else{
+          1
+        }
+      }).reduce((a, b) => a+b)
+      println("TOTAL DEFEATERS: " + defeaterNodes)
+      defeaterNodes
+    }
+
+
+
+    def calcDefeatersMessageCount(sGraph: SocialGraph) : Long = {
+      // Init graph to (Weight, -1, False)
+      println("Preparing environment for calculate message count")
+      val graph = sGraph.getSubGraph(true, false, false).getRepresentation().FloatIntegerLongBool(0, 0, 0, false, true, true)
+      //println("USED VERTICES: " + graph.numVertices)
+      val sssp = graph.pregel((0, -1, 0, false))(
+        (id, main, received) => {
+          //Initial message, avoid mark leafs as unfinished
+          if(received._2 == -1){
+            (main._1, 0, 0, main._4)
+          }
+          else{
+            //If the state it's the same as previous mark as finished
+            (main._1, received._2, main._3 + received._3, main._2 == received._2 || received._4)
+          }
+        },
+        triplet => {
+          //Vote to finish
+          if(triplet.srcAttr._4 && triplet.dstAttr._4){
+            Iterator.empty
+          }
+          else{
+            //Compute the messages to send
+            //Send a message
+            if(triplet.srcAttr._2 == 0 && triplet.srcAttr._1 >= triplet.dstAttr._1){
+              Iterator((triplet.dstId, (0, 1, 1, triplet.srcAttr._4)))
+            }
+            else{
+              Iterator((triplet.dstId, (0, 0, 1, triplet.srcAttr._4)))
+            }
+          }
+        },
+        (a,b) => {
+          //Join messages into one
+          (0, a._2 + b._2, a._3 + b._3, a._4 && b._4)
+        }
+      )
+      val totalMessages = sssp.vertices.reduce((a,b) => (0, (0, 0, a._2._3 + b._2._3, true)))
+      println("TOTAL MESSAGES: " + totalMessages._2._3)
+      val defeaterNodes = sssp.vertices.map(node => {
+        if(node._2._2 > 0){
+          0
+        }
+        else{
+          1
+        }
+      }).reduce((a, b) => a+b)
+      println("TOTAL DEFEATERS: " + defeaterNodes)
+      defeaterNodes
+    }
+
+
+    def optimizedCalcDefeatersMessageCount(sGraph: SocialGraph ) : scala.xml.Elem = {
+      // Graph node state: (Weight, NumAcceptedAttackers, NumReceivedMessages, Finished )
+      //
+      // Init graph to ((Weight,INdegree,DEFsINdegree), (0,0), 0, False) this is the format for main (vertex state)
+      val graph = sGraph.getSubGraph(true, false, false).getRepresentation().FloatIntegerIntegerIntegerLongBool( true, 0, 0, 0, false, true)
+      var solutionlist = sGraph.getListOfDisconnectedNodes()
+
+      val sssp = graph.pregel(initialMsg = (0, (0,0), 0),activeDirection = EdgeDirection.Out)(
+        (id, main, received) => {
+          // Map message "received" by node "id" with current state "main"
+          // return new state with same format as main
+          //
+          //Evaluate the joined messages and set the new state
+          // println( "Node "+id.toString()+" State: "+main.toString()+" Message received: "+received.toString())
+
+          val newaccepteddefeaters = main._2._1 + received._2._1
+          val newrejecteddefeaters = main._2._2 + received._2._2
+          // Check if number of rejected defeaters equals to DEFsINdegree
+          if( (newrejecteddefeaters) ==  main._1._3 ){
+
+            // println(" PRE:"+main._2.toString()+"=CURRENT "+received._2.toString()+", vertex "+id.toString()+ " defeaters: "+received._2.toString()+ " is finished ")
+            (main._1,  (newaccepteddefeaters ,newrejecteddefeaters), main._3 + received._3, true)
+          }
+          else{
+            // Check if it has at least one accepted defeater
+            if ( (newaccepteddefeaters) >= 1){
+              // println(" PRE:"+main._2.toString()+"=CURRENT "+received._2.toString()+", vertex "+id.toString()+ " defeaters: "+received._2.toString()+ " is finished ")
+              (main._1, (newaccepteddefeaters , newrejecteddefeaters), main._3 + received._3, true )
+            }
+            else{
+              // println( "No final state for : " +id.toString()+ " state:   ("+newaccepteddefeaters.toString()+ " " + newrejecteddefeaters.toString() )
+              (main._1, (newaccepteddefeaters , newrejecteddefeaters), main._3 + received._3, false)
+            }
+          }
+        },
+        triplet => {
+          //Compute the message to send for edge edge triplet
+          // triplet.srcAttr, triplet.attr, triplet.dstAttr
+          if(triplet.srcAttr._1._1 >= triplet.dstAttr._1._1 && //Weight Source >= Weight Destination
+             triplet.srcAttr._4 && !triplet.dstAttr._4){       // Src Finished but Dst is not finished
+            //Send a "defeater accepted message" to dstId, if #rejected attackers == INdegree
+            if(triplet.srcAttr._2._2 == triplet.srcAttr._1._3){
+              // If Source node has ALL defeaters rejected, add one more accepted defeater for Dst
+              // println( " MESSAGE: "+ triplet.srcId.toString()+ " -> "+triplet.dstId.toString()+" = (1,0)")
+              Iterator((triplet.dstId, (0, (1,0), 1)))
+            }
+            else { if (triplet.srcAttr._2._1 >= 1) {
+              // If Source node has some accepted defeaters, add one more  rejected defeater for Dst
+              // println( " MESSAGE: "+ triplet.srcId.toString()+ " -> "+triplet.dstId.toString()+" = (0,1)")
+              Iterator((triplet.dstId, (0, (0,1), 1)))
+            }
+            else{
+              //Don't send a message
+              Iterator.empty
+            } }
+          }
+          else{
+            //Don't send a message
+            Iterator.empty
+          }
+        },
+        (a,b) => {
+          //Join messages into one
+          (0, (a._2._1 + b._2._1, a._2._2 + b._2._2), a._3 + b._3 )
+        }
+      )
+      val totalMessages = sssp.vertices.reduce((a,b) => (0, ((0,0,0), (0,0), a._2._3 + b._2._3, true)))
+      println("TOTAL MESSAGES: " + totalMessages._2._3)
+      val defeaterNodes = sssp.vertices.map(node => {
+        if(node._2._2._2 == node._2._1._3) {  1  }  else{  0  }       }
+      ).reduce((a, b) => a+b)
+
+      // node =  (id,((Weight,INdegree), ( accepted,rejected ), messagesrecv, Finished))
+      val listofdefeaterNodes = sssp.vertices.filter( node => (node._2._2._2 == node._2._1._3)).
+         map( node => {node._1.toString() } ).collect()
+      println("total ACCEPTED without disconnected: " + defeaterNodes)
+
+      listofdefeaterNodes.foreach( solutionlist += )
+      println("LIST OF ACCEPTED ONES:" )
+      solutionlist.foreach( println )
+      println("END OF LIST" )
+      println("TOTAL ACCEPTED: "+solutionlist.length.toString())
+      <ArgumentationAnswers><answer>{solutionlist.map(i => <arg id={i} />)}</answer></ArgumentationAnswers>
+    }
+  }

distributed_social_network_analyzer/src/main/scala/Learning/MP/MPClassifier.scala

@@ -0,0 +1,79 @@
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.SparkSession
+import java.io._
+import scala.io._
+import scala.util.matching.Regex
+import scala.collection.mutable.ListBuffer
+
+// Import classes for MLLib
+import org.apache.spark.ml.classification.MultilayerPerceptronClassifier
+import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
+import org.apache.spark.ml.classification.MultilayerPerceptronClassificationModel;
+
+// @Annotation
+// TODO: Re-Think how to classify tweets passed by parameter and how to return
+// the result. Modify directly the file (?)
+
+
+
+
+// MultilayerPerceptronClassifier
+class MPC(ss: SparkSession, input: String,
+                            indexes: String,
+                            model: String){
+
+    // Load indexes
+    // if (input != "") {
+    val text = ss.read.textFile(indexes)
+    val pair_indexes = text.collect().map(parseTweetPair).toList
+    // Load Data
+    val data = ss.read.format("libsvm").load(input)
+    // Load Model
+    val mp_model = MultilayerPerceptronClassificationModel.load(model)
+    // Classifies
+    val results = mp_model.transform(data)
+    // Get Prediction column
+    val prediction_column = results.select("prediction")
+    // Get Predictions
+    var predictions = prediction_column.collect.map(_.getDouble(0).toInt)
+  // }
+
+
+  // def this(ss: SparkSession ) {
+  // this(ss,"","","")
+  // }
+  
+  def getRelationship(id1: Long, id2: Long): String = {
+    var pair = TweetPair(id1.toString, id2.toString)
+    var index = getIndexes(pair, pair_indexes)
+    var label = if (index > 0) predictions(index) else 0
+    return getMeaningOfLabel(label)
+  }
+
+  def getIndexes(pair: TweetPair, indexes: List[TweetPair]): Int = {
+    var i = 0
+    for(index <- indexes){
+      if (index.equals(pair))
+        return i
+      i += 1
+    }
+    return -1
+  }
+
+  def getMeaningOfLabel(relation: Int): String = {
+    val meaning = relation match{
+      case 1 => "attack"
+      case 2 => "support"
+      case _ => "noone"
+    }
+    return meaning
+  }
+
+  case class TweetPair(id1: String, id2: String)
+
+  def parseTweetPair(str: String): TweetPair = {
+    val line = str.split("  ")
+    TweetPair(line(0), line(1))
+  }
+
+}

distributed_social_network_analyzer/src/main/scala/Learning/MP/MPTraining.scala

@@ -0,0 +1,101 @@
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+import scala.util.Random
+
+// Import classes for ML
+import org.apache.spark.ml.linalg.Vector
+import org.apache.spark.ml.classification.MultilayerPerceptronClassifier
+import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
+import org.apache.spark.ml.classification.MultilayerPerceptronClassificationModel;
+
+// MultilayerPerceptronTraining
+class MPT(ss: SparkSession, input: String) {
+
+  // Load the data stored in LIBSVM format as a DataFrame.
+  val data = ss.read.format("libsvm").load(input)
+
+  case class Params(var training_size : Double = 0.75, var maxIters : Int = 100, var seed: Long = 1234L,
+               var block: Int = 128, var layers : Array[Int] = Array.empty[Int])
+
+  def generateModel(in_params: Params, output_file: String = "",
+                    random: Boolean = false): Double = {
+
+
+    // Set Layers
+    val numFeatures = data.first().getAs[Vector](1).size
+    val classes = 3
+
+    var params = in_params
+    setLayer(params, numFeatures, classes)
+
+    if (random)
+      getRandomParams(params)
+
+
+    // Split the data into train and test
+    val splits = data.randomSplit(Array(params.training_size, 1-params.training_size))
+    val (train, test) = (splits(0), splits(1))
+
+    // Generate Model
+    val trainer = new MultilayerPerceptronClassifier()
+      .setLayers(params.layers)
+      .setBlockSize(params.block)
+      .setSeed(params.seed)
+      .setMaxIter(params.maxIters)
+    val model = trainer.fit(train)
+
+    // Save model
+    if(output_file == "")
+      model.write.overwrite().save("target/tmp/MPM")
+    else
+      model.write.overwrite().save(output_file)
+
+    // Get Accuracy
+    val accuracy = getAccuracy(model, test)
+    return accuracy
+    }
+
+    // Set to the layers the features and classes
+    def setLayer(params: Params, num_features : Int, output_classes: Int){
+      params.layers = Array(num_features) ++ params.layers ++ Array(output_classes)
+    }
+
+    // Dummy Sample
+    def getRandomParams(params: Params){
+      var i = 5
+    }
+
+    // Generates random layers
+    def getRandomLayer(num_features : Int, output_classes: Int): Array[Int] = {
+      val r = Random
+      var num_layers = r.nextInt(5)
+      if (num_layers != 0)
+        num_layers += 1
+      val layers = new Array[Int](num_layers + 2)
+
+      //Initial Features
+      layers(0) = num_features
+
+      //Output classes
+      layers(num_layers+1) = output_classes
+
+      for(i <- 1 to num_layers)
+        layers(i) = r.nextInt(50) + 10
+
+      return layers
+    }
+
+    // Returns accuracy of a given model with a data set
+    def getAccuracy(model: MultilayerPerceptronClassificationModel,
+                      test: DataFrame): Double = {
+      //compute accuracy on the test set
+      val result = model.transform(test)
+      val predictionAndLabels = result.select("prediction", "label")
+      val evaluator = new MulticlassClassificationEvaluator()
+        .setMetricName("accuracy")
+
+      val accuracy = evaluator.evaluate(predictionAndLabels)
+      return accuracy
+    }
+}

distributed_social_network_analyzer/src/main/scala/Learning/MachineLearning.scala

@@ -0,0 +1,25 @@
+// import org.apache.spark.sql.SparkSession
+// import org.apache.spark.sql.DataFrame
+// import scala.collection.mutable.ListBuffer
+//
+//
+// object MachineLearning{
+//   def main(args: Array[String]) {
+//     val ss = SparkSession
+//       .builder
+//       .appName("Machine Learning")
+//       .getOrCreate()
+//
+//     //Execute methods
+//     val model = new MPT(ss, args(0))
+//     val accuracy = model.generateModel(new model.Params(), "target/tmp/MPM")
+//     println("DEBUG : ACCURACY => " + accuracy)
+//
+//     val classifier = new MPC(ss, args(0), args(1), "target/tmp/MPM")
+//     var relationShip = classifier.getRelationship(588748014438764544L, 588727284061863936L)
+//     println("DEBUG : LABEL => " + relationShip)
+//
+//     ss.stop()
+//   }
+//
+// }

distributed_social_network_analyzer/src/main/scala/Learning/OtherModels/DecisionTreeModelCustom.scala

@@ -0,0 +1,134 @@
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+
+// Import classes for ML
+import org.apache.spark.ml.{Pipeline,PipelineModel}
+import org.apache.spark.ml.classification.DecisionTreeClassifier
+import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
+import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
+
+// DecisionTreeModelCustom
+object DTMC {
+
+	class Params(var input : String, var output : String, var maxDepth : Int,
+               var maxBins: Int) {
+    def this() {
+        this("", "", 8, 200);
+    }
+		override def toString : String =
+		      "Params: \n" +
+		      " - Input =    " + input + "\n" +
+		      " - Output =   " + output + "\n" +
+		      " - MaxDepth = " + maxDepth + "\n" +
+		      " - MaxBins =  " + maxBins + "\n"
+		  }
+
+	def main(args: Array[String]) {
+		val sc = SparkSession
+      .builder
+      .appName("Decision Tree Model")
+      .getOrCreate()
+
+		//Parse and Save Params
+		val arglist = args.toList
+		val params = new Params()
+		getOptions(arglist, params)
+
+		if(params.input == ""){
+			val msg = """Usage -> arguments: -arg value
+									argument: -input required""".stripMargin
+			println(msg)
+			sys.exit(1)
+		}
+
+		// Load the data stored in LIBSVM format as a DataFrame.
+    val data = sc.read.format("libsvm").load(params.input)
+
+		// Split the data into train and test
+    val splits = data.randomSplit(Array(0.75, 0.25))
+    val (train, test) = (splits(0), splits(1))
+
+		// Index labels, adding metadata to the label column.
+		// Fit on whole dataset to include all labels in index.
+		val labelIndexer = new StringIndexer()
+		  .setInputCol("label")
+		  .setOutputCol("indexedLabel")
+		  .fit(data)
+		// Automatically identify categorical features, and index them.
+		val featureIndexer = new VectorIndexer()
+		  .setInputCol("features")
+		  .setOutputCol("indexedFeatures")
+		  .setMaxCategories(350) // features with > 4 distinct values are treated as continuous.
+		  .fit(data)
+
+		// Train a DecisionTree model.
+		val dt = new DecisionTreeClassifier()
+		  .setLabelCol("indexedLabel")
+		  .setFeaturesCol("indexedFeatures")
+			.setFeaturesCol("indexedFeatures")
+			.setMaxDepth(params.maxDepth)
+			.setMaxBins(params.maxBins)
+			// .setMinInstancesPerNode(params.minInstancesPerNode)
+			// .setMinInfoGain(params.minInfoGain)
+			// .setCacheNodeIds(params.cacheNodeIds)
+			// .setCheckpointInterval(params.checkpointInterval)
+
+		// Convert indexed labels back to original labels.
+		val labelConverter = new IndexToString()
+		  .setInputCol("prediction")
+		  .setOutputCol("predictedLabel")
+		  .setLabels(labelIndexer.labels)
+
+		// Chain indexers and tree in a Pipeline.
+		val pipeline = new Pipeline()
+		  .setStages(Array(labelIndexer, featureIndexer, dt, labelConverter))
+
+		// Train model. This also runs the indexers.
+		val model = pipeline.fit(train)
+
+    // Get Accuracy
+		val accuracy = getAccuracy(model, test)
+		println("Acurracy: " + accuracy)
+
+		// Save model
+    if(params.output == "")
+      model.write.overwrite().save("target/tmp/DTM")
+    else
+      model.write.overwrite().save(params.output)
+
+		sc.stop()
+  	}
+
+		def getOptions(list: List[String], params: Params){
+      def isSwitch(s : String) = (s(0) == '-')
+      list match {
+        case Nil =>
+        case "-input"  :: value :: tail => params.input = value.toString
+                                           getOptions(tail, params)
+        case "-output" :: value :: tail => params.output = value.toString
+                                           getOptions(tail, params)
+        case "-maxd"    :: value :: tail => params.maxDepth = value.toInt
+                                           getOptions(tail, params)
+				case "-maxb"    :: value :: tail => params.maxBins = value.toInt
+																	         getOptions(tail, params)
+        case option :: tail => println("Unknown option " + option)
+                               sys.exit(1)
+      }
+
+    }
+
+		//Auxiliar Function: returns accuracy of a given model with a data set
+    def getAccuracy(model: PipelineModel,
+                      test: DataFrame): Double = {
+      //compute accuracy on the test set
+      val result = model.transform(test)
+      val predictionAndLabels = result.select("prediction", "label")
+      val evaluator = new MulticlassClassificationEvaluator()
+        .setMetricName("accuracy")
+
+      val accuracy = evaluator.evaluate(predictionAndLabels)
+      return accuracy
+    }
+
+}

distributed_social_network_analyzer/src/main/scala/Learning/OtherModels/Examples/ExampleFlights.scala

@@ -0,0 +1,115 @@
+import org.apache.spark.{SparkConf, SparkContext}
+
+import org.apache.spark.rdd.RDD
+
+// Import classes for MLLib
+import org.apache.spark.mllib.regression.LabeledPoint
+import org.apache.spark.mllib.linalg.Vectors
+import org.apache.spark.mllib.tree.DecisionTree
+import org.apache.spark.mllib.tree.model.DecisionTreeModel
+import org.apache.spark.mllib.util.MLUtils
+
+object ExampleFlights {
+
+	def main(args: Array[String]) {
+		val conf = new SparkConf().setAppName("EF")
+		val sc = new SparkContext(conf)
+
+		//Load Data into a RDD, modify route to required
+		val textRDD = sc.textFile("flights.csv")
+
+		//Parse the RDD of csv lines into an RDD of flight classes
+		val flightsRDD = textRDD.map(parseFlight).cache()
+		cPrint(flightsRDD.first().toString)
+
+		// EXTRACT FEATURES: map string city to int
+		var cityMap: Map[String, Int] = Map()
+		var index: Int = 0
+		flightsRDD.map(flight => flight.src).distinct.collect.foreach(
+					   x => { cityMap += (x -> index); index += 1 })
+		cPrint(cityMap.toString)
+
+		// DEFINE FEATURES ARRAY
+		val mlprep = flightsRDD.map(flight => {
+					  val src = cityMap(flight.src) // category
+					  val dest = cityMap(flight.dest) // category
+					  val id = flight.id.toInt
+					  val price = flight.price.toInt
+					  val expensive = if (flight.price.toInt > 500) 1 else 0
+					  Array(expensive.toInt, src.toInt, dest.toInt, id.toInt)
+					})
+		println(mlprep.take(5).deep.mkString("\n"))
+
+		// CREATE LABELED POINTS
+		val mldata = mlprep.map(x => LabeledPoint(x(0), Vectors.dense(x(1), x(2), x(3))))
+		println(mldata.take(5).deep.mkString("\n"))
+
+		// SPLIT DATA: TRAINING SET - TEST SET
+		// Flights with price < 500 -> get 50%
+		val mldata0 = mldata.filter(x => x.label == 0.0).randomSplit(Array(0.50, 0.50))(1)
+		// Flights with price > 500 -> get 50%
+		val mldata1 = mldata.filter(x => x.label == 1.0).randomSplit(Array(0.50, 0.50))(1)
+		// FLights selected
+		//val mldata2 = mldata0 ++ mldata1
+		val mldata2 = mldata
+
+		//Split mldata2 into training and test data
+		val splits = mldata2.randomSplit(Array(0.3, 0.7))
+		val (trainingData, testData) = (splits(0), splits(1))
+
+		// INFO ABOUT FEATURES
+		var categoricalFeaturesInfo = Map[Int, Int]()
+		categoricalFeaturesInfo += (0 -> cityMap.size) //src
+		categoricalFeaturesInfo += (1 -> cityMap.size) //dest
+
+
+		// VALUES OF MODEL
+		val numClasses = 2
+		// Defining values for the other parameters
+		val impurity = "gini"
+		val maxDepth = 10
+		val maxBins = 5000
+
+		// Call DecisionTree trainClassifier with the trainingData , which returns the model
+		val model = DecisionTree.trainClassifier(trainingData, numClasses,
+			categoricalFeaturesInfo, impurity, maxDepth, maxBins)
+		model.toDebugString
+
+		//val labelAndPreds = testData.map { point =>
+		val labelAndPreds = testData.map { point =>
+			val prediction = model.predict(point.features)
+			(point.label, prediction)
+			}
+
+		val goodPrediction =(labelAndPreds.filter{
+			case (label, prediction) => ( label == prediction)
+			})
+		val wrongPrediction =(labelAndPreds.filter{
+			case (label, prediction) => ( label != prediction)
+			})
+		println(labelAndPreds.take(10000).deep.mkString("\n"))
+
+		val good = goodPrediction.count()
+		val wrong = wrongPrediction.count()
+		val ratioWrong=wrong.toDouble/testData.count()
+
+		val results = "Good Prediction : " + good.toString + "\n" +
+					  "Wrong Prediction: " + wrong.toString + "\n" +
+					  "Ratio Wrong     : " + ratioWrong.toString + "\n"
+		cPrint(results)
+
+		sc.stop()
+  	}
+
+	case class Flight(src: String, dest: String, id: Int, price: Int)
+
+	def parseFlight(str: String): Flight = {
+		val line = str.split(",")
+		Flight(line(0), line(1), line(2).toInt, line(3).toInt)
+	}
+
+	def cPrint(str: String) {
+		val output = "\n" + str + "\n\n"
+		print(output)
+	}
+}

distributed_social_network_analyzer/src/main/scala/Learning/OtherModels/NaiveBayesModelCustom.scala

@@ -0,0 +1,111 @@
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+
+// Import classes for ML
+import org.apache.spark.ml.classification.NaiveBayes
+import org.apache.spark.ml.classification.NaiveBayesModel
+import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
+
+object NBMC {
+
+  class Params(var input : String, var output : String, var maxIters : Int,
+               var block: Int, var seed: Long, var layers : Array[Int]) {
+    def this() {
+        this("", "", 100, 128, 1234L, Array[Int]());
+    }
+    override def toString : String =
+      "Params: \n" +
+      " - Input =    " + input + "\n" +
+      " - Output =   " + output + "\n" +
+      " - MaxIters = " + maxIters + "\n" +
+      " - Block =    " + block + "\n" +
+      " - Seed =     " + seed + "\n" +
+      " - Layers =   " + layers + "\n"
+
+  }
+
+  def main(args: Array[String]) {
+    val sc = SparkSession
+      .builder
+      .appName("NaiveBayes Model")
+      .getOrCreate()
+
+    //Parse and Save Params
+    val arglist = args.toList
+    val params = new Params()
+    getOptions(arglist, params)
+
+    if(params.input == ""){
+      val msg = """Usage -> arguments: -arg value
+                  argument: -input required""".stripMargin
+      println(msg)
+      sys.exit(1)
+    }
+
+    // Load the data stored in LIBSVM format as a DataFrame.
+    val data = sc.read.format("libsvm").load(params.input)
+
+    // Split the data into train and test
+    val splits = data.randomSplit(Array(0.75, 0.25))
+    val (train, test) = (splits(0), splits(1))
+
+    // Parse params
+    parseParams(params)
+
+    // Train a NaiveBayes model.
+    val model = new NaiveBayes()
+      .fit(train)
+
+    // Get Accuracy
+    val accuracy = getAccuracy(model, test)
+    printf("Accuracy: " + accuracy + "\n")
+
+    // Save model
+    if(params.output == "")
+      model.write.overwrite().save("target/tmp/MPM")
+    else
+      model.write.overwrite().save(params.output)
+
+    sc.stop()
+    }
+
+    def getOptions(list: List[String], params: Params){
+      def isSwitch(s : String) = (s(0) == '-')
+      list match {
+        case Nil =>
+        case "-input"  :: value :: tail => params.input = value.toString
+                                           getOptions(tail, params)
+        case "-output" :: value :: tail => params.output = value.toString
+                                           getOptions(tail, params)
+        case "-max"    :: value :: tail => params.maxIters = value.toInt
+                                           getOptions(tail, params)
+        case "-block"  :: value :: tail => params.block = value.toInt
+                                           getOptions(tail, params)
+        case "-seed"   :: value :: tail => params.seed = value.toLong
+                                           getOptions(tail, params)
+        case option :: tail => println("Unknown option " + option)
+                               sys.exit(1)
+      }
+
+    }
+
+    def parseParams(params: Params){
+      //seed
+      params.seed = System.currentTimeMillis()
+    }
+
+
+    //Auxiliar Function: returns accuracy of a given model with a data set
+    def getAccuracy(model: NaiveBayesModel,
+                      test: DataFrame): Double = {
+      //compute accuracy on the test set
+      val result = model.transform(test)
+      val predictionAndLabels = result.select("prediction", "label")
+      val evaluator = new MulticlassClassificationEvaluator()
+        .setMetricName("accuracy")
+
+      val accuracy = evaluator.evaluate(predictionAndLabels)
+      return accuracy
+    }
+}

distributed_social_network_analyzer/src/main/scala/MainApp.scala

@@ -0,0 +1,37 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+import scala.collection.mutable.ListBuffer
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+
+object MainApp{
+  def main(args: Array[String]){
+    val ss = SparkSession
+      .builder
+      .appName("Main Application")
+      .getOrCreate()
+    val sc = ss.sparkContext
+    val path = "Storage/Jsons/" + args(0) + ".json"
+    //Create MPC
+    val parserCaller = new ParserCaller(ss)
+    val classifier = parserCaller.parseJson(args(0))
+    //Create graph
+    val graph = new TwitterGraph(sc, classifier)
+    graph.parseConversationFromFile(path)
+    val calcs : RelationshipsCalcs = new RelationshipsCalcs()
+    calcs.optimizedCalcDefeatersMessageCount(graph)
+    ss.stop()
+  }
+}

distributed_social_network_analyzer/src/main/scala/MainAppFromXML.scala

@@ -0,0 +1,49 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+import scala.collection.mutable.ListBuffer
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+import org.apache.log4j._
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+
+import scala.xml.XML
+
+// <ArgumentationAnswers ArgumentationProblemInstance=name
+//   Semantics="VAFideal"><answer><arg id="588729571383574528" /></answer>></ArgumentationAnswers>
+
+
+object MainAppFromXML{
+  def main(args: Array[String]){
+    val ss = SparkSession
+      .builder
+      .appName("Main Application")
+      .getOrCreate()
+    val sc = ss.sparkContext
+    // <ArgumentationAnswers><answer> </answer></ArgumentationAnswers>
+
+    LogManager.getLogger("org").setLevel(Level.OFF)
+    val path = args(0) + ".xml"
+    //Create MPC
+    // val parserCaller = new ParserCaller(ss)
+    // val classifier = parserCaller.parseJson(args(0))
+    // val nullclassifier = new MPC(ss)
+    //Create graph
+    val graph = new TwitterGraphFromXML( sc )
+    graph.parseXMLLabelledConversation( path )
+    val calcs : RelationshipsCalcs = new RelationshipsCalcs()
+    val xmlsolution = calcs.optimizedCalcDefeatersMessageCount(graph)
+    ss.stop()
+    XML.save(args(0)+"-xml.sol",xmlsolution)
+  }
+}

distributed_social_network_analyzer/src/main/scala/ParserCaller/ParserCaller.scala

@@ -0,0 +1,22 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+import scala.collection.mutable.ListBuffer
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+class ParserCaller (session: SparkSession){
+  def parseJson(id: String) : MPC = {
+    val context = session.sparkContext
+    val indexes = "Storage/Files/pairs-" + id + ".txt"
+    val input = "Storage/Files/attributes-" + id + ".txt"
+    val model = "Storage/Models/MPM"
+    // val scriptPath = "parser/parser_all.py"
+    // val scriptName = "parser_all.py"
+    // context.addFile(scriptPath)
+    // val data = context.parallelize(List(json_file))
+    // val pipeRDD = data.pipe(scriptPath)
+    // pipeRDD.collect()
+    return new MPC(session, input, indexes, model)
+  }
+}

distributed_social_network_analyzer/src/main/scala/Tests/BasicTests.scala

@@ -0,0 +1,226 @@
+//SparkContext & Session
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+
+class BasicTests(session: SparkSession){
+
+  val context = session.sparkContext
+
+  def doAllTests(){
+    graphParsingTest()
+    graphCreatingTest()
+    parseSingleTweetTest()
+    parseTweetConversation()
+    multireferenceTweetConversation()
+  }
+
+  def graphParsingTest(){
+    val jGraph = new SocialGraph(context)
+
+    //Make the connections, both systems are equivalent
+
+    jGraph.parseConnection("{\"origin\":1," +
+                            "\"destination\":2," +
+                            "\"intention\":\"noone\"}")
+    jGraph.parseConnection("{\"origin\":1," +
+                            "\"destination\":3," +
+                            "\"intention\":\"noone\"}")
+
+    //Make the messages identifier, author and Json with properties)
+
+    jGraph.parseMessage("{\"id\":1," +
+                         "\"weight\":1," +
+                         "\"author\":\"author1\"," +
+                         "\"message\":\"content1\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+    jGraph.parseMessage("{\"id\":2," +
+                         "\"weight\":1," +
+                         "\"author\":\"author2\"," +
+                         "\"message\":\"content1\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+    jGraph.parseMessage("{\"id\":3," +
+                         "\"weight\":1," +
+                         "\"author\":\"author3\"," +
+                         "\"message\":\"content2\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+
+    jGraph.makeGraph()
+
+    val graph = jGraph.getGraph()
+
+    val contentCount = graph.vertices.filter{
+      case(id, message) => message.message == "content1"
+    }.count
+
+    val relationToRootCount = graph.edges.filter(e => e.srcId < e.dstId).count
+
+
+    println("Total messages with content 1: " + contentCount + "\n" +
+      "Total edges going to higher node id: " + relationToRootCount)
+  }
+
+  def listSubGraphTriplets(){
+    val jGraph = new SocialGraph(context)
+    jGraph.addConnection(2, 1, "attack")
+    jGraph.addConnection(3, 2, "support")
+    jGraph.addConnection(4, 1, "support")
+    jGraph.addConnection(5, 4, "attack")
+    jGraph.addConnection(6, 4, "noone")
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    jGraph.addMessage(1, "author1", "content1", someMap, 1)
+    jGraph.addMessage(2, "author2", "content2", someMap, 2)
+    jGraph.addMessage(3, "author3", "content3", someMap, 3)
+    jGraph.addMessage(4, "author4", "content4", someMap, 1)
+    jGraph.addMessage(5, "author5", "content5", someMap, 2)
+    jGraph.addMessage(6, "author6", "content6", someMap, 3)
+
+    val graph = jGraph.getSubGraph(true, false, false).getGraph()
+    val facts: RDD[String] =
+    graph.triplets.map(triplet =>
+    triplet.srcAttr.author + " is " + triplet.attr + " with " + triplet.dstAttr.author)
+    facts.collect.foreach(println(_))
+  }
+
+  def graphCreatingTest(){
+    val jGraph = new SocialGraph(context)
+    jGraph.addConnection(1, 2, "noone")
+    jGraph.addConnection(1, 3, "noone")
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    jGraph.addMessage(1, "author1", "content1", someMap, 1)
+    jGraph.addMessage(2, "author2", "content2", someMap, 2)
+    jGraph.addMessage(3, "author3", "content3", someMap, 3)
+
+    jGraph.makeGraph()
+
+    val graph = jGraph.getGraph()
+
+    val contentCount = graph.vertices.filter{
+      case(id, message) => message.message == "content1"
+    }.count
+
+    val relationToRootCount = graph.edges.filter(e => e.srcId < e.dstId).count
+
+
+    println("Total messages with content 1: " + contentCount + "\n" +
+      "Total edges going to higher node id: " + relationToRootCount)
+  }
+
+  def parseSingleTweetTest(){
+
+    //Create MPC
+    val path = "Examples/JsonSingleTweet.txt"
+    val parserCaller = new ParserCaller(session)
+    val classifier = parserCaller.parseJson(path)
+    //Create graph
+    val tGraph = new TwitterGraph(context, classifier)
+    tGraph.parseConversationFromFile(path)
+    val line = Source.fromFile(path).getLines().next()
+    println("Let's parse the next tweet: \n" + line)
+    tGraph.parseTweet(line)
+  }
+
+  def parseTweetConversation(){
+    //Create MPC
+    val path = "Examples/JsonConversationTweets.txt"
+    val parserCaller = new ParserCaller(session)
+    val classifier = parserCaller.parseJson(path)
+    //Create graph
+    val tGraph = new TwitterGraph(context, classifier)
+    tGraph.parseConversationFromFile(path)
+  }
+
+  def multireferenceTweetConversation(){
+    println("[TEST]: Let's parse MultiReferenceConversation")
+
+    //Create MPC
+    val path = "Examples/JsonMultireferenceConversation.txt"
+    val parserCaller = new ParserCaller(session)
+    val classifier = parserCaller.parseJson(path)
+    //Create graph
+    val tGraph = new TwitterGraph(context, classifier)
+    tGraph.parseConversationFromFile(path)
+  }
+
+  def parseRedditConversation(){
+    println("[TEST]: Let's parse Reddit conversation from file")
+    val rGraph = new RedditGraph(context)
+    val filePath = "Examples/FakeRedditConversation.txt"
+    rGraph.parseRedditConversationFromFile(filePath)
+  }
+
+  def updatePunctuation(){
+
+    val jGraph = new SocialGraph(context)
+
+    //Make the connections, both systems are equivalent
+
+    jGraph.parseConnection("{\"origin\":1," +
+                            "\"destination\":2," +
+                            "\"intention\":\"noone\"}")
+    jGraph.parseConnection("{\"origin\":1," +
+                            "\"destination\":3," +
+                            "\"intention\":\"noone\"}")
+
+    //Make the messages identifier, author and Json with properties)
+
+    jGraph.parseMessage("{\"id\":1," +
+                         "\"weight\":1," +
+                         "\"author\":\"author1\"," +
+                         "\"message\":\"content1\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+    jGraph.parseMessage("{\"id\":2," +
+                         "\"weight\":1," +
+                         "\"author\":\"author2\"," +
+                         "\"message\":\"content1\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+    jGraph.parseMessage("{\"id\":3," +
+                         "\"weight\":1," +
+                         "\"author\":\"author3\"," +
+                         "\"message\":\"content2\"," +
+                         "\"properties\":{\"property\":\"something\"}}")
+
+    jGraph.makeGraph()
+
+    val graph = jGraph.getGraph()
+
+    val contentCount = graph.vertices.filter{
+      case(id, message) => message.message == "content1"
+    }.count
+
+    val relationToRootCount = graph.edges.filter(e => e.srcId < e.dstId).count
+
+
+    println("Total messages with content 1: " + contentCount + "\n" +
+      "Total edges going to higher node id: " + relationToRootCount)
+
+    val facts: RDD[String] =
+    graph.triplets.map(triplet =>
+    triplet.srcAttr.author + " with " + triplet.srcAttr.weight + " is " + triplet.attr + " with " + triplet.dstAttr.author + " with " + triplet.dstAttr.weight)
+
+    //Update values
+    jGraph.updatePunctuation(1, 4)
+    jGraph.makeGraph()
+    val newFacts: RDD[String] =
+    jGraph.getGraph().triplets.map(triplet =>
+    triplet.srcAttr.author + " with " + triplet.srcAttr.weight + " is " + triplet.attr + " with " + triplet.dstAttr.author + " with " + triplet.dstAttr.weight)
+
+    println("Old values...")
+    facts.collect.foreach(println(_))
+    println("New values...")
+    newFacts.collect.foreach(println(_))
+
+  }
+
+}

distributed_social_network_analyzer/src/main/scala/Tests/DistributedGraphTests.scala

@@ -0,0 +1,185 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+import sys.process._
+import java.io._
+
+class DistributedGraphTests (session: SparkSession){
+
+  val context = session.sparkContext
+
+  def listTriplets(){
+    val testGraph : TwitterGraph = getMultiReferenceConversationGraph()
+    val graph = testGraph.getGraph()
+    val facts: RDD[String] =
+    graph.triplets.map(triplet =>
+    triplet.srcAttr.author + " is " + triplet.attr + " with " + triplet.dstAttr.author)
+    facts.collect.foreach(println(_))
+  }
+
+  def twitterDistanceTest(){
+    println("[DISTRIBUTED_TEST]: Pregel test")
+    val testGraph : TwitterGraph = getMultiReferenceConversationGraph()
+    println("Max distance of the graph is " + testGraph.maxDistance()
+         +"\nMin distance of the graph is " + testGraph.minDistance()
+         +"\nAverage distance is " + testGraph.averageDistance())
+  }
+
+  def distancesTest(){
+    //Create a graph with min depth 2 and max depth 3
+    val sGraph : SocialGraph = new SocialGraph(context)
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    sGraph.addMessage(1, "author1", "content1", someMap)
+    sGraph.addMessage(2, "author2", "content2", someMap)
+    sGraph.addMessage(3, "author3", "content3", someMap)
+    sGraph.addMessage(4, "author4", "content4", someMap)
+    sGraph.addMessage(5, "author5", "content5", someMap)
+    sGraph.addMessage(6, "author6", "content6", someMap)
+    sGraph.addMessage(7, "author7", "content7", someMap)
+    sGraph.addMessage(8, "author8", "content8", someMap)
+
+    sGraph.addConnection(2, 1, "noone")
+    sGraph.addConnection(3, 1, "noone")
+    sGraph.addConnection(4, 2, "noone")
+    sGraph.addConnection(5, 2, "noone")
+    sGraph.addConnection(6, 3, "noone")
+    sGraph.addConnection(7, 3, "noone")
+    sGraph.addConnection(8, 5, "noone")
+
+    println("Min distance of the graph is " + sGraph.minDistance()
+         +"\nMax distance of the graph is " + sGraph.maxDistance()
+         +"\nAverage distance is " + sGraph.averageDistance())
+  }
+
+  def distancesTestAlt(){
+    //Create a graph with min depth 2 and max depth 3
+    val sGraph : SocialGraph = new SocialGraph(context)
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    sGraph.addMessage(1,  "author1",  "content1",  someMap)
+    sGraph.addMessage(2,  "author2",  "content2",  someMap)
+    sGraph.addMessage(3,  "author3",  "content3",  someMap)
+    sGraph.addMessage(4,  "author4",  "content4",  someMap)
+    sGraph.addMessage(5,  "author5",  "content5",  someMap)
+    sGraph.addMessage(6,  "author6",  "content6",  someMap)
+    sGraph.addMessage(7,  "author7",  "content7",  someMap)
+    sGraph.addMessage(8,  "author8",  "content8",  someMap)
+    sGraph.addMessage(9,  "author9",  "content9",  someMap)
+    sGraph.addMessage(10, "author10", "content10", someMap)
+
+    sGraph.addConnection(2,  1, "noone")
+    sGraph.addConnection(3,  1, "noone")
+    sGraph.addConnection(4,  2, "noone")
+    sGraph.addConnection(5,  2, "noone")
+    sGraph.addConnection(6,  2, "noone")
+    sGraph.addConnection(7,  5, "noone")
+    sGraph.addConnection(8,  5, "noone")
+    sGraph.addConnection(9,  3, "noone")
+    sGraph.addConnection(10, 3, "noone")
+
+
+    println("Min distance of the graph is " + sGraph.minDistance()
+         +"\nMax distance of the graph is " + sGraph.maxDistance()
+         +"\nAverage distance is " + sGraph.averageDistance())
+  }
+
+  def defeatersCountTest(){
+    //Create a graph with min depth 2 and max depth 3
+    val sGraph : SocialGraph = new SocialGraph(context)
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    sGraph.addMessage(1,  "author1",  "content1",  someMap)
+    sGraph.addMessage(2,  "author2",  "content2",  someMap)
+    sGraph.addMessage(3,  "author3",  "content3",  someMap)
+    sGraph.addMessage(4,  "author4",  "content4",  someMap)
+    sGraph.addMessage(5,  "author5",  "content5",  someMap)
+    sGraph.addMessage(6,  "author6",  "content6",  someMap)
+    sGraph.addMessage(7,  "author7",  "content7",  someMap)
+    sGraph.addMessage(8,  "author8",  "content8",  someMap)
+    sGraph.addMessage(9,  "author9",  "content9",  someMap)
+    sGraph.addMessage(10, "author10", "content10", someMap)
+
+    sGraph.addConnection(2,  1, "noone")
+    sGraph.addConnection(3,  1, "noone")
+    sGraph.addConnection(4,  2, "noone")
+    sGraph.addConnection(5,  2, "noone")
+    sGraph.addConnection(6,  2, "noone")
+    sGraph.addConnection(7,  5, "noone")
+    sGraph.addConnection(8,  5, "noone")
+    sGraph.addConnection(9,  3, "noone")
+    sGraph.addConnection(10, 3, "noone")
+
+    val calcs: RelationshipsCalcs = new RelationshipsCalcs()
+    calcs.calcDefeaters(sGraph)
+  }
+
+  def optimizedDefeatersCountTest(){
+    //Create a graph with min depth 2 and max depth 3
+    val sGraph : SocialGraph = new SocialGraph(context)
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    sGraph.addMessage(1,  "author1",  "content1",  someMap)
+    sGraph.addMessage(2,  "author2",  "content2",  someMap)
+    sGraph.addMessage(3,  "author3",  "content3",  someMap)
+    sGraph.addMessage(4,  "author4",  "content4",  someMap)
+    sGraph.addMessage(5,  "author5",  "content5",  someMap)
+    sGraph.addMessage(6,  "author6",  "content6",  someMap)
+    sGraph.addMessage(7,  "author7",  "content7",  someMap)
+    sGraph.addMessage(8,  "author8",  "content8",  someMap)
+    sGraph.addMessage(9,  "author9",  "content9",  someMap)
+    sGraph.addMessage(10, "author10", "content10", someMap)
+
+    sGraph.addConnection(2,  1, "noone")
+    sGraph.addConnection(3,  1, "noone")
+    sGraph.addConnection(4,  2, "noone")
+    sGraph.addConnection(5,  2, "noone")
+    sGraph.addConnection(6,  2, "noone")
+    sGraph.addConnection(7,  5, "noone")
+    sGraph.addConnection(8,  5, "noone")
+    sGraph.addConnection(9,  3, "noone")
+    sGraph.addConnection(10, 3, "noone")
+
+    val calcs: RelationshipsCalcs = new RelationshipsCalcs()
+    calcs.optimizedCalcDefeaters(sGraph)
+  }
+
+  def downloadConversation(){
+    val testGraph : TwitterGraph = getMultiReferenceConversationGraph()
+  }
+
+  def downloadRedditConversation(){
+    val downloader = new RedditDownloader(context)
+    val testGraph : RedditGraph = downloader.getGraphFromConversationId("63w4sq","Downloaders/reddit-collector/reddit.json")
+  }
+
+  private def getMultiReferenceConversationGraph(): TwitterGraph = {
+    storedTestGraph match {
+      case Some(i) => i
+      case None =>  {
+        val downloader = new TwitterDownloader(session.sparkContext)
+        val path = downloader.downloadAndStoreConversation("574324656905281538")
+    
+        //Create MPC
+        val parserCaller = new ParserCaller(session)
+        val classifier = parserCaller.parseJson(path)
+        //Create graph
+        val testGraph = new TwitterGraph(session.sparkContext, classifier)
+        testGraph.parseConversationFromFile(path)
+        storedTestGraph = Some(testGraph)
+        testGraph
+      }
+    }
+  }
+
+  var storedTestGraph : Option[TwitterGraph] = None
+
+}

distributed_social_network_analyzer/src/main/scala/Tests/GraphMPCTests.scala

@@ -0,0 +1,46 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.DataFrame
+import scala.collection.mutable.ListBuffer
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+import sys.process._
+import java.io._
+
+class GraphMPCTests (session: SparkSession){
+
+  def listTriplets(id: String){
+    val downloader = new TwitterDownloader(session.sparkContext)
+    val path = downloader.downloadAndStoreConversation(id)
+
+    //Create MPC
+    val parserCaller = new ParserCaller(session)
+    val classifier = parserCaller.parseJson(path)
+    //Create graph
+    val testGraph = new TwitterGraph(session.sparkContext, classifier)
+    testGraph.parseConversationFromFile(path)
+    val graph = testGraph.getGraph()
+    val facts: RDD[String] =
+    graph.triplets.map(triplet =>
+    triplet.srcAttr.author + " is " + triplet.attr + " with " + triplet.dstAttr.author)
+    facts.collect.foreach(println(_))
+  }
+
+  /*
+   Some conversations:
+   835171254852202496
+
+  */
+
+}

distributed_social_network_analyzer/src/main/scala/Tests/SerializationTests.scala

@@ -0,0 +1,100 @@
+//SparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.SparkContext
+import org.apache.spark.SparkConf
+
+//GraphX TODO: TEST ONLY
+import org.apache.spark._
+import org.apache.spark.graphx._
+import org.apache.spark.rdd.RDD
+
+//Lists
+import scala.collection.mutable.ListBuffer
+
+//Read files
+import scala.io.Source
+
+class SerializationTests(session: SparkSession){
+
+  val context = session.sparkContext
+
+  def graphSerailizingTest(){
+    val jGraph = new SocialGraph(context)
+    jGraph.addConnection(1, 2, "noone")
+    jGraph.addConnection(1, 3, "noone")
+    val someMap : Map[String, Any] = Map("Something" -> "Content")
+    jGraph.addMessage(1, "author1", "content1", someMap)
+    jGraph.addMessage(2, "author2", "content2", someMap)
+    jGraph.addMessage(3, "author3", "content3", someMap)
+
+    jGraph.makeGraph()
+
+    val graph = jGraph.getGraph()
+
+    val contentCount = graph.vertices.filter{
+      case(id, message) => message.message == "content1"
+    }.count
+
+    val relationToRootCount = graph.edges.filter(e => e.srcId < e.dstId).count
+
+
+    println("Total messages with content 1: " + contentCount + "\n" +
+      "Total edges going to higher node id: " + relationToRootCount)
+    jGraph.storeGraph("Storage/Graphs/", "Test1")
+
+    val lGraph = new SocialGraph(context)
+    lGraph.loadGraph("Storage/Graphs/", "Test1")
+    lGraph.makeGraph()
+
+    val loaded_graph = lGraph.getGraph()
+
+    val loaded_contentCount = loaded_graph.vertices.filter{
+      case(id, message) => message.message == "content1"
+    }.count
+
+    val loaded_relationToRootCount = loaded_graph.edges.filter(e => e.srcId < e.dstId).count
+
+
+    println("[LOADED GRAPH]Total messages with content 1: " + loaded_contentCount + "\n" +
+      "[LOADED GRAPH]Total edges going to higher node id: " + loaded_relationToRootCount)
+  }
+
+  def twitterStoringTest(){
+    println("[SERIALIZATION_TEST]: Storing Twitter graph")
+    val testGraph : TwitterGraph = getMultiReferenceConversationGraph()
+    println("[SERIALIZATION_TEST]: Twitter graph downloaded, storging...")
+    testGraph.storeGraph("Storage/Graphs/", "TwitterTest")
+    println("[SERIALIZATION_TEST]: stored, loading from storage")
+    val lGraph = new SocialGraph(context)
+    lGraph.loadGraph("Storage/Graphs/", "TwitterTest")
+    lGraph.makeGraph()
+
+    println("Max distance of the graph is " + testGraph.maxDistance()
+         +"\nMin distance of the graph is " + testGraph.minDistance()
+         +"\nAverage distance is " + testGraph.averageDistance())
+   println("Max distance of the LOADED graph is " + lGraph.maxDistance()
+        +"\nMin distance of the LOADED graph is " + lGraph.minDistance()
+        +"\nAverage distance LOADED is " + lGraph.averageDistance())
+  }
+
+  private def getMultiReferenceConversationGraph(): TwitterGraph = {
+    storedTestGraph match {
+      case Some(i) => i
+      case None =>  {
+        val downloader = new TwitterDownloader(session.sparkContext)
+        val path = downloader.downloadAndStoreConversation("574324656905281538")
+        //Create MPC
+        val parserCaller = new ParserCaller(session)
+        val classifier = parserCaller.parseJson(path)
+        //Create graph
+        val testGraph = new TwitterGraph(session.sparkContext, classifier)
+        testGraph.parseConversationFromFile(path)
+        storedTestGraph = Some(testGraph)
+        testGraph
+      }
+    }
+  }
+
+  var storedTestGraph : Option[TwitterGraph] = None
+
+}

smallexamples/Deguindos-c5-logweights.xml

@@ -0,0 +1,480 @@
+<entailment-corpus>
+  <argument-list maxweight="5" minweight="3">
+    <arg emoticon="none" hashlist="[]" id="590480494636179456" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Marxista_esp en 2014, 24% y en 2015 23%</arg>
+    <arg emoticon="none" hashlist="[]" id="590480675385507840" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo y estamos en 2016? @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590480949516840960" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Marxista_esp s&#237; hemos estado en 2014 (24%) y s&#237; estamos en 2015 (23%)</arg>
+    <arg emoticon="none" hashlist="[]" id="590484317673037824" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo si, pero si se aplica a ejercicio 2015 se paga en 2016 @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590484586448211968" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Marxista_esp Pero ya se est&#225; devengando.</arg>
+    <arg emoticon="none" hashlist="[]" id="590484707697172480" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo si, 4 meses. Eso explica la historia impositiva de Dinamarca @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590484974027075584" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Marxista_esp Explica que se maltrata menos a la empresas que aqu&#237;.</arg>
+    <arg emoticon="none" hashlist="[]" id="590485435106906112" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo el tipo reducido en Espa&#241;a es un 25% y tb se va a bajar @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590485945889251328" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Marxista_esp En Dinamarca el no reducido es el 24%</arg>
+    <arg emoticon="none" hashlist="[]" id="590486764147576832" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo sabe usted que en Espa&#241;a se puede pagar un 10% de sociedades sin contar deducciones? @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590487344848961536" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @juanrallo @Marxista_esp e incluso un 5% afirmado por el mismisimo DeGuindos! :O</arg>
+    <arg emoticon="none" hashlist="[]" id="590487439002710016" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Ilias_a @Marxista_esp Qu&#233; fe tienen en mentirosos profesionales!</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590487744603893761" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo Si no te crees a un Ministro ... Apaga y vamonos :) habra que hablar con un Experto Fiscaliza @Remo_ ;)  @Ilias_a @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590488016088657921" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Remo_ @Ilias_a @Marxista_esp Jaja. Tambi&#233;n te cre&#237;as a Rato supongo.</arg>
+    <arg emoticon="none" hashlist="[]" id="590488286591848448" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo cuando daba datos fiscales del Gobierno no es para creerlo? @Remo_ @Ilias_a @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590489251113992192" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Remo_ @Ilias_a @Marxista_esp El gobiernos tiene inter&#233;s en maximizar los impuestos. Es parte interesada.</arg>
+    <arg emoticon="none" hashlist="[]" id="590489338514907136" lang="en" numhashs="0" numurls="1" sentiment="0" topic="conv-c5-Deguindos" urllist="[u'http://ec.europa.eu/eurostat/tgm/table.do?tab=table&amp;init=1&amp;language=en&amp;pcode=tec00020&amp;plugin=1']" weight="3">@TrabajarEuropa @juanrallo Taxes on production and imports SPA 11% DEN 16% http://t.co/aJ8a1BdIYL @Remo_ @Marxista_esp @TrabajarEuropa</arg>
+    <arg emoticon="none" hashlist="[]" id="590489516974153728" lang="es" numhashs="0" numurls="0" sentiment="-2" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo no, si detr&#225;s del gobierno hay organizaciones privadas haciendo lobby  @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590489610930761728" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp Jaja, eso son los impuestos indirectos tipo IVA. En fin, qu&#233; nivel.</arg>
+    <arg emoticon="none" hashlist="[]" id="590490256740392960" lang="es" numhashs="0" numurls="1" sentiment="-1" topic="conv-c5-Deguindos" urllist="[u'http://ec.europa.eu/eurostat/tgm/web/table/description.jsp']" weight="3">@juanrallo perd&#243;n pero en la definici&#243;n no dice eso http://t.co/hSjHgg5x9w @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590490593370988545" lang="es" numhashs="0" numurls="1" sentiment="-1" topic="conv-c5-Deguindos" urllist="[u'https://stats.oecd.org/glossary/detail.asp?ID=2670']" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp Impuestos sobre bienes cuando son vendidos? https://t.co/JJ6dvxbiyD no sabe qu&#233; es el IVA?</arg>
+    <arg emoticon="none" hashlist="[]" id="590490652137357312" lang="en" numhashs="0" numurls="0" sentiment="2" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo employment of labour, the ownership or use of land, buildings or other assets used in production @TrabajarEuropa @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590490781351288832" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo IVA es un impuesto al consumo, no a la producci&#243;n @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590490886334771200" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo y no creo q el impuesto al trabajo, o al uso de tierra sea IVA @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491046469095424" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Marxista_esp el impuesto de Sociedades es un impuesto sobre el trabajo?</arg>
+    <arg emoticon="none" hashlist="[]" id="590491264816144384" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp Es sobre el Valor A&#241;adido. El Valor A&#241;adido de producci&#243;n. En fin. Nivel.</arg>
+    <arg emoticon="none" hashlist="[]" id="590491281186562048" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo le remito a la parte inicial del debate donde usted dec&#237;a q Dinamarca s&#243;lo paga IRPF e IVA @TrabajarEuropa @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491359095685122" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp Lo que desde luego no es, es Sociedades.</arg>
+    <arg emoticon="none" hashlist="[]" id="590491426976354304" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo no al trabajo, pero si un impuesto sobre la capacidad productiva @TrabajarEuropa @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491512514969600" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Marxista_esp Ah bien, que ahora tratamos de escurrir el bulto.</arg>
+    <arg emoticon="none" hashlist="[]" id="590491561814839296" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo si, pero lo paga el consumidor. No es un impuesto a la producci&#243;n @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491619222233088" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="4">@TrabajarEuropa @juanrallo @Ilias_a @Marxista_esp un 10% sin deducciones ni ajustes? Me lo expliquen</arg>
+    <arg emoticon="none" hashlist="[]" id="590491664621432833" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo el IVA es un impuesto al consumo,@TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491690491850752" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Marxista_esp Falso. Es un impuesto sobre beneficios. Sobre la RENTA neta empresarial.</arg>
+    <arg emoticon="none" hashlist="[]" id="590491854099128320" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo Dinamarca no es Canarias ? eso lo habr&#225; hecho usted con los del 1pp  entre DIN y ESP @TrabajarEuropa @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590491950526156800" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp Jaja. Qu&#233; tendr&#225; que ver quien sea el obligado tributario con cu&#225;l sea la base imponible?</arg>
+    <arg emoticon="none" hashlist="[]" id="590492057027903489" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp un impuesto sobre la producci&#243;n que paga el consumidor.</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590492219255169024" lang="es" medialist="[u'http://twitter.com/TrabajarEuropa/status/590492219255169024/photo/1']" numhashs="0" nummedia="1" numurls="0" sentiment="1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @juanrallo @Marxista_esp en fin... Seria este grafico... Supongo :) http://t.co/g3kLWKqx3k</arg>
+    <arg emoticon="none" hashlist="[]" id="590492230248435712" lang="es" numhashs="0" numurls="1" sentiment="0" topic="conv-c5-Deguindos" urllist="[u'http://trueeconomics.blogspot.com.es/2011/06/20062011-europes-corporate-tax-rates.html?m=1']" weight="5">@Ilias_a @TrabajarEuropa @Marxista_esp Much&#237;simo m&#225;s en tipos efectivos: http://t.co/911L3fbTM1</arg>
+    <arg emoticon="none" hashlist="[]" id="590492347827429376" lang="es" numhashs="0" numurls="0" sentiment="-2" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo no , no es as&#237; @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590492417540939776" lang="es" numhashs="0" numurls="0" sentiment="1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Ilias_a @Marxista_esp y d&#243;nde sale Dinamarca en ese gr&#225;fico? :)</arg>
+    <arg emoticon="none" hashlist="[]" id="590492470498177024" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo quien paga el IVA no es la empresa, es el consumidor @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590492660193976320" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo jeje es que ahi dice que en ESP es del 40!!! Jajaja imagina... :-P @Ilias_a @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590492672751771649" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp qu&#233; es el valor a&#241;adido?</arg>
+    <arg emoticon="none" hashlist="[]" id="590492798199160832" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp jaja. Y d&#243;nde dice que los impuestos sobre la producci&#243;n los pague la empresa?Se est&#225; luciendo</arg>
+    <arg emoticon="none" hashlist="[]" id="590492915761356801" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Ilias_a @Marxista_esp Ah. Ah&#237; ya no se cree a Guindos? Qu&#233; cosas.</arg>
+    <arg emoticon="none" hashlist="[]" id="590493032157421569" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo jejeje, pero los costes laborales y los impuestos s/ la tierra a producir suele ser que si @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590493253461475328" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @TrabajarEuropa @Remo_ @Marxista_esp O no. El IRPF es un impuesto sobre las rentas salariales y no lo paga la empresa.</arg>
+    <arg emoticon="none" hashlist="[]" id="590493330997452800" lang="es" numhashs="0" numurls="0" sentiment="-3" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo no es que no me lo crea... Es que han metido ahi todo tipo de cosas que no son SOLO el tipo sobre Bcios @Ilias_a @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590493498979328000" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Ilias_a @Marxista_esp y si no le gusta ese gr&#225;fico, para qu&#233; me lo env&#237;a? Ruido?</arg>
+    <arg emoticon="none" hashlist="[]" id="590493617128665089" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo en todo caso que cosas... Somos el 11 con tasas totales MAS BAJAS! :-P jajaja @Ilias_a @Marxista_esp</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590493729057812480" lang="es" numhashs="0" numurls="0" sentiment="1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@juanrallo pues porque me confundi! :)</arg>
+    <arg emoticon="none" hashlist="[]" id="590493831356882945" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa @Ilias_a @Marxista_esp tasas? Mezcla impuestos con tasas? Y discut&#237;amos Espa&#241;a vs. Dinamarca.</arg>
+    <arg emoticon="none" hashlist="[]" id="590494029277716480" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@TrabajarEuropa Ah ya, que lo envi&#243; antes de fijarse d&#243;nde estaba Dinamarca.</arg>
+    <arg emoticon="none" hashlist="[u'EU']" id="590495047709229056" lang="es" numhashs="1" numurls="1" sentiment="0" topic="conv-c5-Deguindos" urllist="[u'http://ec.europa.eu/taxation_customs/resources/documents/common/publications/studies/final_report_2014_taxud_2013_cc_120.pdf']" weight="5">@juanrallo @Ilias_a @Marxista_esp bueno... Si quereis ver tipo EFECTIVO en #EU bucear por las mas de 3.00 paginas de http://t.co/wtvlxhNIOu</arg>
+    <arg emoticon="none" hashlist="[]" id="590496429631746048" lang="es" medialist="[u'http://twitter.com/TrabajarEuropa/status/590496429631746048/photo/1']" numhashs="0" nummedia="1" numurls="1" sentiment="-1" topic="conv-c5-Deguindos" urllist="[u'http://ec.europa.eu/transparency/regexpert/index.cfm?do=groupDetail.groupDetailDoc&amp;id=11838&amp;no=3']" weight="5">@juanrallo @Ilias_a @Marxista_esp lo unico que he encontrado es este graf. http://t.co/l4W2pK8mAJ pero no esta Denn http://t.co/Qeq2ZZbVni</arg>
+    <arg emoticon="none" hashlist="[]" id="590497227090571264" lang="es" medialist="[u'http://twitter.com/Ilias_a/status/590497227090571264/photo/1']" numhashs="0" nummedia="1" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo mire esta tabla, OCDE imp s/ b&#186; de las corporaciones en % del PIB DIN 3,3% @TrabajarEuropa @Marxista_esp http://t.co/Auhalax6Ey</arg>
+    <arg emoticon="none" hashlist="[]" id="590497405122048000" lang="es" medialist="[u'http://twitter.com/Ilias_a/status/590497405122048000/photo/1']" numhashs="0" nummedia="1" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo y ahora la misma tabla pero ESP= 2% seguimos con lo de jaja, jeje, jiji ?@TrabajarEuropa @Marxista_esp http://t.co/t6lSecAJIK</arg>
+    <arg emoticon="none" hashlist="[]" id="590497633510289408" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a pero te dira que eso depende del PIb... Que el quiere el tipo "oficial" jaja 
+
+No hay ningun oriundo de DK que nos pueda informar?</arg>
+    <arg emoticon="none" hashlist="[]" id="590498344830644225" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa es un soci&#243;pata narcisista.... ese es el problema. Lo de Dinamarca no es Canarias fue el colmo</arg>
+    <arg emoticon="none" hashlist="[]" id="590498484077338624" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa y fue el &#250;nico argumento para no responder ante lo del RIC</arg>
+    <arg emoticon="none" hashlist="[]" id="590498573499944960" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa porque en Canarias las empresas q se acogen a la RIC s&#243;lo pagan un 10% de sociedades</arg>
+    <arg emoticon="none" hashlist="[]" id="590501865936318464" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ pues con la RIC , pero tienes q estar en Canarias @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590502238566670337" lang="es" numhashs="0" numurls="1" sentiment="0" topic="conv-c5-Deguindos" urllist="[u'http://www.emprenderencanarias.es/ref-intro/reserva-inversiones/']" weight="3">@Remo_  aqu&#237; est&#225; http://t.co/XUtP3k31I7 @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590502313799868416" lang="es" numhashs="0" numurls="0" sentiment="2" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ y ha ido como el culo @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590502788964216832" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ mira la enorme fortaleza de la econom&#237;a canaria desde q se aplica la RIC (iron&#237;a) @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590502934393315328" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ nuestras multinacionales dominan el mundo... (iron&#237;a) @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590505662846791680" lang="es" numhashs="0" numurls="1" sentiment="1" topic="conv-c5-Deguindos" urllist="[u'http://elblogdedaniel.com/2015/02/14/mito-1-las-grandes-empresas-solo-tributan-un-3/']" weight="3">@TrabajarEuropa @Ilias_a @juanrallo @Marxista_esp si te interesa,escribo sobre el mito del "3%" :) http://t.co/h2doz5GksA</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590507061496123392" lang="es" numhashs="0" numurls="0" sentiment="1" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Daniel_Marbella gracias! Me lo apunto! A ver si indica que pagan el 25% :)  @Ilias_a @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590508470773936128" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa que bueno, el art&#237;culo se lo pasa La Calle...  @Daniel_Marbella @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590512258591105024" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="5">@Ilias_a @Daniel_Marbella huis.. El LaCalle que me tiene bloqueao?? :-P jajaja</arg>
+    <arg emoticon="none" hashlist="[]" id="590516047951835136" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa atenci&#243;n al siguiente zasca.....</arg>
+    <arg emoticon="none" hashlist="[]" id="590517879948369920" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@juanrallo tiene q ver mucho porque no se tiene q vender donde se produce  @TrabajarEuropa @Remo_ @Marxista_esp</arg>
+    <arg emoticon="positive" emoticon-positive-counter="1" hashlist="[]" id="590518373307572225" lang="es" numhashs="0" numurls="0" sentiment="1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@TrabajarEuropa @Ilias_a @juanrallo @Marxista_esp gracias por apuntartelo :)</arg>
+    <arg emoticon="none" hashlist="[]" id="590538494688231425" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="4">@Ilias_a trampas al solitario las justas, que en Melilla tambi&#233;n se pagan muy pocos impuestos @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590539899171897345" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ si, pero Canarias son 2 millones de habitantes y no veo donde hay trampa @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590540065576787971" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ un impuesto de sociedades del 10% en territorio espa&#241;ol , que trampa? @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590540394649276417" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ y posiblemente Dinamarca con poblaci&#243;n d 5 millones sea + comparable a a 2 millones q a 45 @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590540582637989889" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ y cualquier empresa del territorio espa&#241;ol puede colocar su sede fiscal donde quiera @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590543705792208896" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="4">@Ilias_a falso, el domicilio fiscal tiene unas reglas muy claras y no se puede poner donde quieras @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590543815460704258" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="4">@Ilias_a digo donde quieras si tu centro efectivo o mayor inmovilizado est&#225; fuera de Canarias @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590544220643061760" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ si, pero puedes poner una empresa v&#237;nculo y expatriar beneficios @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590544581214744576" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ como operan las grandes con los para&#237;sos fiscales o Irlanda @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590544652060733441" lang="es" numhashs="0" numurls="0" sentiment="-1" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ por tanto no es falso @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590544955967463427" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="4">@Ilias_a tampoco, aplica el art&#237;culo 18 de la LIS para operaciones vinculadas. @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590546008968396800" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ en ese art&#237;culo se habla del valor contable y valor de mercado?? que tiene que ver? 
+@TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590546999075856384" lang="es" numhashs="0" numurls="0" sentiment="-2" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ estar&#225;s refiriendote al art&#237;culo 8, y en el no dice q no se pueda dentro de Espa&#241;a @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590547200545058816" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ y que yo sepa, de momento Canarias est&#225; dentro de Espa&#241;a @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590547374365376512" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ lo que tu dices vale para operar desde un paraiso fiscal fuera de territorio espa&#241;ol @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590556202477707265" lang="es" numhashs="0" numurls="1" sentiment="0" topic="conv-c5-Deguindos" urllist="[u'http://noticias.juridicas.com/base_datos/Fiscal/540273-l-27-2014-de-27-nov-impuesto-sobre-sociedades.html#a18']" weight="4">@Ilias_a no, me refiero a esto http://t.co/jnL3KyeRGW y la filial Canaria que planteas @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590571044727054338" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ si tienes una filial en Canarias, puedes introducir ingresos de varias formas @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590571312965427200" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ como servicios realizados por la filial Canaria @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590571540925784064" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ a ver, hay empresas q ya lo hacen @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590572057961857024" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ de hecho trabaj&#233; en una empresa donde lo hac&#237;amos al rev&#233;s desde Canarias a Barcelona @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+    <arg emoticon="none" hashlist="[]" id="590572306466009090" lang="es" numhashs="0" numurls="0" sentiment="0" topic="conv-c5-Deguindos" urllist="[]" weight="3">@Remo_ en que parte del territorio imputas la actividad d una empresa q produce x toda Esp? @TrabajarEuropa @juanrallo @Marxista_esp</arg>
+  </argument-list>
+  <argument-pairs>
+    <pair entailment="ATTACKS" id="0" topic="conv-c5-Deguindos">
+      <t id="590480675385507840"/>
+      <h id="590480494636179456"/>
+    </pair>
+    <pair entailment="ATTACKS" id="1" topic="conv-c5-Deguindos">
+      <t id="590491359095685122"/>
+      <h id="590490886334771200"/>
+    </pair>
+    <pair entailment="ATTACKS" id="2" topic="conv-c5-Deguindos">
+      <t id="590540065576787971"/>
+      <h id="590538494688231425"/>
+    </pair>
+    <pair entailment="ATTACKS" id="3" topic="conv-c5-Deguindos">
+      <t id="590487439002710016"/>
+      <h id="590487344848961536"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="4" topic="conv-c5-Deguindos">
+      <t id="590487344848961536"/>
+      <h id="590486764147576832"/>
+    </pair>
+    <pair entailment="ATTACKS" id="5" topic="conv-c5-Deguindos">
+      <t id="590488016088657921"/>
+      <h id="590487744603893761"/>
+    </pair>
+    <pair entailment="ATTACKS" id="6" topic="conv-c5-Deguindos">
+      <t id="590543815460704258"/>
+      <h id="590543705792208896"/>
+    </pair>
+    <pair entailment="ATTACKS" id="7" topic="conv-c5-Deguindos">
+      <t id="590544955967463427"/>
+      <h id="590544220643061760"/>
+    </pair>
+    <pair entailment="ATTACKS" id="8" topic="conv-c5-Deguindos">
+      <t id="590488286591848448"/>
+      <h id="590488016088657921"/>
+    </pair>
+    <pair entailment="ATTACKS" id="9" topic="conv-c5-Deguindos">
+      <t id="590538494688231425"/>
+      <h id="590501865936318464"/>
+    </pair>
+    <pair entailment="ATTACKS" id="10" topic="conv-c5-Deguindos">
+      <t id="590497633510289408"/>
+      <h id="590497227090571264"/>
+    </pair>
+    <pair entailment="ATTACKS" id="11" topic="conv-c5-Deguindos">
+      <t id="590492347827429376"/>
+      <h id="590492057027903489"/>
+    </pair>
+    <pair entailment="ATTACKS" id="12" topic="conv-c5-Deguindos">
+      <t id="590539899171897345"/>
+      <h id="590538494688231425"/>
+    </pair>
+    <pair entailment="ATTACKS" id="13" topic="conv-c5-Deguindos">
+      <t id="590497405122048000"/>
+      <h id="590492417540939776"/>
+    </pair>
+    <pair entailment="ATTACKS" id="14" topic="conv-c5-Deguindos">
+      <t id="590492660193976320"/>
+      <h id="590492417540939776"/>
+    </pair>
+    <pair entailment="ATTACKS" id="15" topic="conv-c5-Deguindos">
+      <t id="590546008968396800"/>
+      <h id="590544955967463427"/>
+    </pair>
+    <pair entailment="ATTACKS" id="16" topic="conv-c5-Deguindos">
+      <t id="590495047709229056"/>
+      <h id="590493831356882945"/>
+    </pair>
+    <pair entailment="ATTACKS" id="17" topic="conv-c5-Deguindos">
+      <t id="590491561814839296"/>
+      <h id="590491264816144384"/>
+    </pair>
+    <pair entailment="ATTACKS" id="18" topic="conv-c5-Deguindos">
+      <t id="590492057027903489"/>
+      <h id="590491664621432833"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="19" topic="conv-c5-Deguindos">
+      <t id="590512258591105024"/>
+      <h id="590508470773936128"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="20" topic="conv-c5-Deguindos">
+      <t id="590507061496123392"/>
+      <h id="590505662846791680"/>
+    </pair>
+    <pair entailment="ATTACKS" id="21" topic="conv-c5-Deguindos">
+      <t id="590492417540939776"/>
+      <h id="590492219255169024"/>
+    </pair>
+    <pair entailment="ATTACKS" id="22" topic="conv-c5-Deguindos">
+      <t id="590497227090571264"/>
+      <h id="590492417540939776"/>
+    </pair>
+    <pair entailment="ATTACKS" id="23" topic="conv-c5-Deguindos">
+      <t id="590490781351288832"/>
+      <h id="590489610930761728"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="24" topic="conv-c5-Deguindos">
+      <t id="590505662846791680"/>
+      <h id="590487344848961536"/>
+    </pair>
+    <pair entailment="ATTACKS" id="25" topic="conv-c5-Deguindos">
+      <t id="590498484077338624"/>
+      <h id="590497633510289408"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="26" topic="conv-c5-Deguindos">
+      <t id="590516047951835136"/>
+      <h id="590512258591105024"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="27" topic="conv-c5-Deguindos">
+      <t id="590486764147576832"/>
+      <h id="590485945889251328"/>
+    </pair>
+    <pair entailment="ATTACKS" id="28" topic="conv-c5-Deguindos">
+      <t id="590572057961857024"/>
+      <h id="590556202477707265"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="29" topic="conv-c5-Deguindos">
+      <t id="590540394649276417"/>
+      <h id="590538494688231425"/>
+    </pair>
+    <pair entailment="ATTACKS" id="30" topic="conv-c5-Deguindos">
+      <t id="590491619222233088"/>
+      <h id="590487744603893761"/>
+    </pair>
+    <pair entailment="ATTACKS" id="31" topic="conv-c5-Deguindos">
+      <t id="590490593370988545"/>
+      <h id="590490256740392960"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="32" topic="conv-c5-Deguindos">
+      <t id="590502313799868416"/>
+      <h id="590491619222233088"/>
+    </pair>
+    <pair entailment="ATTACKS" id="33" topic="conv-c5-Deguindos">
+      <t id="590546999075856384"/>
+      <h id="590544955967463427"/>
+    </pair>
+    <pair entailment="ATTACKS" id="34" topic="conv-c5-Deguindos">
+      <t id="590492470498177024"/>
+      <h id="590492057027903489"/>
+    </pair>
+    <pair entailment="ATTACKS" id="35" topic="conv-c5-Deguindos">
+      <t id="590491426976354304"/>
+      <h id="590491046469095424"/>
+    </pair>
+    <pair entailment="ATTACKS" id="36" topic="conv-c5-Deguindos">
+      <t id="590490256740392960"/>
+      <h id="590489610930761728"/>
+    </pair>
+    <pair entailment="ATTACKS" id="37" topic="conv-c5-Deguindos">
+      <t id="590543705792208896"/>
+      <h id="590540582637989889"/>
+    </pair>
+    <pair entailment="ATTACKS" id="38" topic="conv-c5-Deguindos">
+      <t id="590547200545058816"/>
+      <h id="590544955967463427"/>
+    </pair>
+    <pair entailment="ATTACKS" id="39" topic="conv-c5-Deguindos">
+      <t id="590489251113992192"/>
+      <h id="590488286591848448"/>
+    </pair>
+    <pair entailment="ATTACKS" id="40" topic="conv-c5-Deguindos">
+      <t id="590485435106906112"/>
+      <h id="590484974027075584"/>
+    </pair>
+    <pair entailment="ATTACKS" id="41" topic="conv-c5-Deguindos">
+      <t id="590484586448211968"/>
+      <h id="590484317673037824"/>
+    </pair>
+    <pair entailment="NONE" id="42" topic="conv-c5-Deguindos">
+      <t id="590502238566670337"/>
+      <h id="590491619222233088"/>
+    </pair>
+    <pair entailment="ATTACKS" id="43" topic="conv-c5-Deguindos">
+      <t id="590502934393315328"/>
+      <h id="590502788964216832"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="44" topic="conv-c5-Deguindos">
+      <t id="590540582637989889"/>
+      <h id="590538494688231425"/>
+    </pair>
+    <pair entailment="ATTACKS" id="45" topic="conv-c5-Deguindos">
+      <t id="590491281186562048"/>
+      <h id="590491046469095424"/>
+    </pair>
+    <pair entailment="ATTACKS" id="46" topic="conv-c5-Deguindos">
+      <t id="590547374365376512"/>
+      <h id="590544955967463427"/>
+    </pair>
+    <pair entailment="ATTACKS" id="47" topic="conv-c5-Deguindos">
+      <t id="590489338514907136"/>
+      <h id="590488286591848448"/>
+    </pair>
+    <pair entailment="ATTACKS" id="48" topic="conv-c5-Deguindos">
+      <t id="590544652060733441"/>
+      <h id="590543815460704258"/>
+    </pair>
+    <pair entailment="ATTACKS" id="49" topic="conv-c5-Deguindos">
+      <t id="590502788964216832"/>
+      <h id="590491619222233088"/>
+    </pair>
+    <pair entailment="ATTACKS" id="50" topic="conv-c5-Deguindos">
+      <t id="590490886334771200"/>
+      <h id="590489610930761728"/>
+    </pair>
+    <pair entailment="ATTACKS" id="51" topic="conv-c5-Deguindos">
+      <t id="590491690491850752"/>
+      <h id="590491426976354304"/>
+    </pair>
+    <pair entailment="ATTACKS" id="52" topic="conv-c5-Deguindos">
+      <t id="590493729057812480"/>
+      <h id="590493498979328000"/>
+    </pair>
+    <pair entailment="ATTACKS" id="53" topic="conv-c5-Deguindos">
+      <t id="590493498979328000"/>
+      <h id="590493330997452800"/>
+    </pair>
+    <pair entailment="ATTACKS" id="54" topic="conv-c5-Deguindos">
+      <t id="590492672751771649"/>
+      <h id="590492347827429376"/>
+    </pair>
+    <pair entailment="ATTACKS" id="55" topic="conv-c5-Deguindos">
+      <t id="590493831356882945"/>
+      <h id="590493617128665089"/>
+    </pair>
+    <pair entailment="ATTACKS" id="56" topic="conv-c5-Deguindos">
+      <t id="590496429631746048"/>
+      <h id="590493831356882945"/>
+    </pair>
+    <pair entailment="ATTACKS" id="57" topic="conv-c5-Deguindos">
+      <t id="590484317673037824"/>
+      <h id="590480949516840960"/>
+    </pair>
+    <pair entailment="ATTACKS" id="58" topic="conv-c5-Deguindos">
+      <t id="590489610930761728"/>
+      <h id="590489338514907136"/>
+    </pair>
+    <pair entailment="ATTACKS" id="59" topic="conv-c5-Deguindos">
+      <t id="590493032157421569"/>
+      <h id="590492798199160832"/>
+    </pair>
+    <pair entailment="ATTACKS" id="60" topic="conv-c5-Deguindos">
+      <t id="590491046469095424"/>
+      <h id="590490652137357312"/>
+    </pair>
+    <pair entailment="ATTACKS" id="61" topic="conv-c5-Deguindos">
+      <t id="590493330997452800"/>
+      <h id="590492915761356801"/>
+    </pair>
+    <pair entailment="ATTACKS" id="62" topic="conv-c5-Deguindos">
+      <t id="590571044727054338"/>
+      <h id="590556202477707265"/>
+    </pair>
+    <pair entailment="ATTACKS" id="63" topic="conv-c5-Deguindos">
+      <t id="590491950526156800"/>
+      <h id="590491561814839296"/>
+    </pair>
+    <pair entailment="ATTACKS" id="64" topic="conv-c5-Deguindos">
+      <t id="590494029277716480"/>
+      <h id="590493729057812480"/>
+    </pair>
+    <pair entailment="ATTACKS" id="65" topic="conv-c5-Deguindos">
+      <t id="590493253461475328"/>
+      <h id="590493032157421569"/>
+    </pair>
+    <pair entailment="ATTACKS" id="66" topic="conv-c5-Deguindos">
+      <t id="590501865936318464"/>
+      <h id="590491619222233088"/>
+    </pair>
+    <pair entailment="ATTACKS" id="67" topic="conv-c5-Deguindos">
+      <t id="590491854099128320"/>
+      <h id="590491512514969600"/>
+    </pair>
+    <pair entailment="ATTACKS" id="68" topic="conv-c5-Deguindos">
+      <t id="590485945889251328"/>
+      <h id="590485435106906112"/>
+    </pair>
+    <pair entailment="ATTACKS" id="69" topic="conv-c5-Deguindos">
+      <t id="590491264816144384"/>
+      <h id="590490781351288832"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="70" topic="conv-c5-Deguindos">
+      <t id="590518373307572225"/>
+      <h id="590507061496123392"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="71" topic="conv-c5-Deguindos">
+      <t id="590508470773936128"/>
+      <h id="590507061496123392"/>
+    </pair>
+    <pair entailment="ATTACKS" id="72" topic="conv-c5-Deguindos">
+      <t id="590492915761356801"/>
+      <h id="590492660193976320"/>
+    </pair>
+    <pair entailment="ATTACKS" id="73" topic="conv-c5-Deguindos">
+      <t id="590480949516840960"/>
+      <h id="590480675385507840"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="74" topic="conv-c5-Deguindos">
+      <t id="590489516974153728"/>
+      <h id="590489251113992192"/>
+    </pair>
+    <pair entailment="ATTACKS" id="75" topic="conv-c5-Deguindos">
+      <t id="590498573499944960"/>
+      <h id="590497633510289408"/>
+    </pair>
+    <pair entailment="ATTACKS" id="76" topic="conv-c5-Deguindos">
+      <t id="590487744603893761"/>
+      <h id="590487439002710016"/>
+    </pair>
+    <pair entailment="ATTACKS" id="77" topic="conv-c5-Deguindos">
+      <t id="590571540925784064"/>
+      <h id="590556202477707265"/>
+    </pair>
+    <pair entailment="ATTACKS" id="78" topic="conv-c5-Deguindos">
+      <t id="590556202477707265"/>
+      <h id="590547374365376512"/>
+    </pair>
+    <pair entailment="ATTACKS" id="79" topic="conv-c5-Deguindos">
+      <t id="590484707697172480"/>
+      <h id="590484586448211968"/>
+    </pair>
+    <pair entailment="ATTACKS" id="80" topic="conv-c5-Deguindos">
+      <t id="590571312965427200"/>
+      <h id="590556202477707265"/>
+    </pair>
+    <pair entailment="ATTACKS" id="81" topic="conv-c5-Deguindos">
+      <t id="590491512514969600"/>
+      <h id="590491281186562048"/>
+    </pair>
+    <pair entailment="ATTACKS" id="82" topic="conv-c5-Deguindos">
+      <t id="590492798199160832"/>
+      <h id="590492470498177024"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="83" topic="conv-c5-Deguindos">
+      <t id="590498344830644225"/>
+      <h id="590497633510289408"/>
+    </pair>
+    <pair entailment="ATTACKS" id="84" topic="conv-c5-Deguindos">
+      <t id="590572306466009090"/>
+      <h id="590556202477707265"/>
+    </pair>
+    <pair entailment="ATTACKS" id="85" topic="conv-c5-Deguindos">
+      <t id="590490652137357312"/>
+      <h id="590489610930761728"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="86" topic="conv-c5-Deguindos">
+      <t id="590544581214744576"/>
+      <h id="590543815460704258"/>
+    </pair>
+    <pair entailment="ATTACKS" id="87" topic="conv-c5-Deguindos">
+      <t id="590492230248435712"/>
+      <h id="590491854099128320"/>
+    </pair>
+    <pair entailment="ATTACKS" id="88" topic="conv-c5-Deguindos">
+      <t id="590517879948369920"/>
+      <h id="590491950526156800"/>
+    </pair>
+    <pair entailment="ATTACKS" id="89" topic="conv-c5-Deguindos">
+      <t id="590484974027075584"/>
+      <h id="590484707697172480"/>
+    </pair>
+    <pair entailment="ATTACKS" id="90" topic="conv-c5-Deguindos">
+      <t id="590492219255169024"/>
+      <h id="590491281186562048"/>
+    </pair>
+    <pair entailment="SUPPORTS" id="91" topic="conv-c5-Deguindos">
+      <t id="590493617128665089"/>
+      <h id="590492915761356801"/>
+    </pair>
+    <pair entailment="ATTACKS" id="92" topic="conv-c5-Deguindos">
+      <t id="590544220643061760"/>
+      <h id="590543815460704258"/>
+    </pair>
+    <pair entailment="ATTACKS" id="93" topic="conv-c5-Deguindos">
+      <t id="590491664621432833"/>
+      <h id="590491264816144384"/>
+    </pair>
+  </argument-pairs>
+</entailment-corpus>