{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/40444866","id":40444866,"diff_hunk":"@@ -0,0 +1,900 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.flink.ml.classification\n+\n+import java.{lang, util}\n+\n+import org.apache.flink.api.common.functions._\n+import org.apache.flink.api.scala._\n+import org.apache.flink.configuration.Configuration\n+import org.apache.flink.core.fs.FileSystem.WriteMode\n+import org.apache.flink.ml.common.{ParameterMap, Parameter}\n+import org.apache.flink.ml.pipeline.{PredictDataSetOperation, FitOperation, Predictor}\n+import org.apache.flink.util.Collector\n+\n+import scala.collection.JavaConverters._\n+import scala.collection.mutable\n+import scala.collection.mutable.ListBuffer\n+import scala.collection.mutable.Map\n+\n+/**\n+ * While building the model different approaches need to be compared.\n+ * For that purpose the fitParameters are used. Every possibility that might enhance\n+ * the implementation can be chosen separately by using the following list of parameters:\n+ *\n+ * Possibility 1: way of calculating document count\n+ *  P1 = 0 -> use .count() to get count of all documents\n+ *  P1 = 1 -> use a reducer and a mapper to create a broadcast data set containing the count of\n+ *    all documents\n+ *\n+ * Possibility 2: all words in class (order of operators)\n+ *    If p2 = 1 improves the speed, many other calculations must switch their operators, too.\n+ *  P2 = 0 -> first the reducer, than the mapper\n+ *  P2 = 1 -> first the mapper, than the reducer\n+ *\n+ * Possibility 3: way of calculating pwc\n+ *  P2 = 0 -> join singleWordsInClass and allWordsInClass to wordsInClass data set\n+ *  P3 = 1 -> work on singleWordsInClass data set and broadcast allWordsInClass data set\n+ *\n+ * Schneider/Rennie 1: ignore/reduce word frequency information\n+ *  SR1 = 0 -> word frequency information is not ignored\n+ *  SR1 = 1 -> word frequency information is ignored (Schneiders approach)\n+ *  SR1 = 2 -> word frequency information is reduced (Rennies approach)\n+ *\n+ * Schneider1: ignore P(c_j) in cMAP formula\n+ *  S1 = 0 -> normal cMAP formula\n+ *  S2 = 1 -> cMAP without P(c_j)\n+ *\n+ * Rennie1: transform document frequency\n+ *  R1 = 0 -> normal formula\n+ *  R1 = 1 -> apply inverse document frequecy\n+ * Note: if R1 = 1 and SR1 = 2, both approaches get applied.\n+ *\n+ */\n+class MultinomialNaiveBayes extends Predictor[MultinomialNaiveBayes] {\n+\n+  import MultinomialNaiveBayes._\n+\n+  //The model, that stores all needed information that are related to one specific word\n+  var wordRelatedModelData: Option[DataSet[(String, String, Double)]] =\n+    None // (class name -> word -> log P(w|c))\n+\n+  //The model, that stores all needed information that are related to one specifc class+\n+  var classRelatedModelData: Option[DataSet[(String, Double, Double)]] =\n+    None // (class name -> p(c) -> log p(w|c) not in class)\n+\n+  //A data set that stores additional needed information for some of the improvements\n+  var improvementData: Option[DataSet[(String, Double)]] =\n+    None // (word -> log number of documents in all classes / word frequency in all classes\n+\n+  // ============================== Parameter configuration ========================================\n+\n+  def setP1(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(P1, value)\n+    this\n+  }\n+\n+  def setP2(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(P2, value)\n+    this\n+  }\n+\n+  def setP3(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(P3, value)\n+    this\n+  }\n+\n+  def setSR1(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(SR1, value)\n+    this\n+  }\n+\n+  def setS1(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(S1, value)\n+    this\n+  }\n+\n+  def setR1(value: Int): MultinomialNaiveBayes = {\n+    parameters.add(R1, value)\n+    this\n+  }\n+\n+  // =============================================== Methods =======================================\n+\n+  /**\n+   * Save already existing model data created by the NaiveBayes algorithm. Requires the designated\n+   * locations. The saved data is a representation of the [[wordRelatedModelData]] and\n+   * [[classRelatedModelData]].\n+   * @param wordRelated, the save location for the wordRelated data\n+   * @param classRelated, the save location for the classRelated data\n+   */\n+  def saveModelDataSet(wordRelated: String, classRelated: String) : Unit = {\n+    wordRelatedModelData.get.writeAsCsv(wordRelated, \"\\n\", \"|\", WriteMode.OVERWRITE)\n+    classRelatedModelData.get.writeAsCsv(classRelated, \"\\n\", \"|\", WriteMode.OVERWRITE)\n+  }\n+\n+  /**\n+   * Save the improvment data set. Requires the designated save location. The saved data is a\n+   * representation of the [[improvementData]] data set.\n+   * @param path, the save location for the improvment data\n+   */\n+  def saveImprovementDataSet(path: String) : Unit = {\n+    improvementData.get.writeAsCsv(path, \"\\n\", \"|\", WriteMode.OVERWRITE)\n+  }\n+\n+  /**\n+   * Sets the [[wordRelatedModelData]] and the [[classRelatedModelData]] to the given data sets.\n+   * @param wordRelated, the data set representing the wordRelated model\n+   * @param classRelated, the data set representing the classRelated model\n+   */\n+  def setModelDataSet(wordRelated : DataSet[(String, String, Double)],\n+                      classRelated: DataSet[(String, Double, Double)]) : Unit = {\n+    this.wordRelatedModelData = Some(wordRelated)\n+    this.classRelatedModelData = Some(classRelated)\n+  }\n+\n+  def setImprovementDataSet(impSet : DataSet[(String, Double)]) : Unit = {\n+    this.improvementData = Some(impSet)\n+  }\n+\n+}\n+\n+object MultinomialNaiveBayes {\n+\n+  // ========================================== Parameters =========================================\n+  case object P1 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  case object P2 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  case object P3 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  case object SR1 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  case object S1 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  case object R1 extends Parameter[Int] {\n+    override val defaultValue: Option[Int] = Some(0)\n+  }\n+\n+  // ======================================== Factory Methods ======================================\n+  def apply(): MultinomialNaiveBayes = {\n+    new MultinomialNaiveBayes()\n+  }\n+\n+  // ====================================== Operations =============================================\n+  /**\n+   * Trains the models to fit the training data. The resulting\n+   * [[MultinomialNaiveBayes.wordRelatedModelData]] and\n+   * [[MultinomialNaiveBayes.classRelatedModelData]] are stored in the [[MultinomialNaiveBayes]]\n+   * instance.\n+   */\n+\n+  implicit val fitNNB = new FitOperation[MultinomialNaiveBayes, (String, String)] {\n+    /**\n+     * The [[FitOperation]] used to create the model. Requires an instance of\n+     * [[MultinomialNaiveBayes]], a [[ParameterMap]] and the input data set. This data set\n+     * maps (string -> string) containing (label -> text, words separated by \",\")\n+     * @param instance of [[MultinomialNaiveBayes]]\n+     * @param fitParameters, additional parameters\n+     * @param input, the to processed data set\n+     */\n+    override def fit(instance: MultinomialNaiveBayes,\n+                     fitParameters: ParameterMap,\n+                     input: DataSet[(String, String)]): Unit = {\n+\n+      val resultingParameters = instance.parameters ++ fitParameters\n+\n+      //Count the amount of documents for each class.\n+      // 1. Map: replace the document text by a 1\n+      // 2. Group-Reduce: sum the 1s by class\n+      val documentsPerClass: DataSet[(String, Int)] = input.map { input => (input._1, 1)}\n+        .groupBy(0).sum(1) // (class name -> count of documents)\n+\n+      //Count the amount of occurrences of each word for each class.\n+      // 1. FlatMap: split the document into its words and add a 1 to each tuple\n+      // 2. Group-Reduce: sum the 1s by class, word\n+      var singleWordsInClass: DataSet[(String, String, Int)] = input\n+        .flatMap(new SingleWordSplitter())\n+        .groupBy(0, 1).sum(2) // (class name -> word -> count of that word)\n+\n+      //POSSIBILITY 2: all words in class (order of operators)\n+      //SCHNEIDER/RENNIE 1: ignore/reduce word frequency information\n+        //the allWordsInClass data set does only contain distinct\n+        //words for schneiders approach: ndw(cj), nothing changes for rennies approach\n+\n+      val p2 = resultingParameters(P2)\n+\n+      val sr1 = resultingParameters(SR1)\n+\n+      var allWordsInClass: DataSet[(String, Int)] =\n+        null // (class name -> count of all words in that class)\n+\n+      if (p2 == 0) {\n+        if (sr1 == 0 || sr1 == 2) {\n+          //Count all the words for each class.\n+          // 1. Reduce: add the count for each word in a class together\n+          // 2. Map: remove the field that contains the word\n+          allWordsInClass = singleWordsInClass.groupBy(0).reduce {\n+            (singleWords1, singleWords2) =>\n+              (singleWords1._1, singleWords1._2, singleWords1._3 + singleWords2._3)\n+          }.map(singleWords =>\n+            (singleWords._1, singleWords._3)) // (class name -> count of all words in that class)\n+        } else if (sr1 == 1) {\n+          //Count all distinct words for each class.\n+          // 1. Map: set the word count to 1\n+          // 2. Reduce: add the count for each word in a class together\n+          // 3. Map: remove the field that contains the word\n+          allWordsInClass = singleWordsInClass\n+            .map(singleWords => (singleWords._1, singleWords._2, 1))\n+            .groupBy(0).reduce {\n+            (singleWords1, singleWords2) =>\n+              (singleWords1._1, singleWords1._2, singleWords1._3 + singleWords2._3)\n+          }.map(singleWords =>\n+            (singleWords._1, singleWords._3))//(class name -> count of distinct words in that class)\n+        }\n+      } else if (p2 == 1) {\n+        if (sr1 == 0 || sr1 == 2) {\n+          //Count all the words for each class.\n+          // 1. Map: remove the field that contains the word\n+          // 2. Reduce: add the count for each word in a class together\n+          allWordsInClass = singleWordsInClass.map(singleWords => (singleWords._1, singleWords._3))\n+            .groupBy(0).reduce {\n+            (singleWords1, singleWords2) => (singleWords1._1, singleWords1._2 + singleWords2._2)\n+          } // (class name -> count of all words in that class)\n+        } else if (sr1 == 1) {\n+          //Count all distinct words for each class.\n+          // 1. Map: remove the field that contains the word, set the word count to 1\n+          // 2. Reduce: add the count for each word in a class together\n+          allWordsInClass = singleWordsInClass.map(singleWords => (singleWords._1, 1))\n+            .groupBy(0).reduce {\n+            (singleWords1, singleWords2) => (singleWords1._1, singleWords1._2 + singleWords2._2)\n+          } // (class name -> count of distinct words in that class)\n+        }\n+\n+      }\n+\n+      //END SCHNEIDER/RENNIE 1\n+      //END POSSIBILITY 2\n+\n+      //POSSIBILITY 1: way of calculating document count\n+      val p1 = resultingParameters(P1)\n+\n+      var pc: DataSet[(String, Double)] = null // (class name -> P(c) in class)\n+\n+      if (p1 == 0) {\n+        val documentsCount: Double = input.count() //count of all documents\n+        //Calculate P(c)\n+        // 1. Map: divide count of documents for a class through total count of documents\n+        pc = documentsPerClass.map(line => (line._1, line._2 / documentsCount))\n+\n+      } else if (p1 == 1) {\n+        //Create a data set that contains only one double value: the count of all documents\n+        // 1. Reduce: At the count of documents together\n+        // 2. Map: Remove field that contains document identifier\n+        val documentCount: DataSet[(Double)] = documentsPerClass\n+          .reduce((line1, line2) => (line1._1, line1._2 + line2._2))\n+          .map(line => line._2) //(count of all documents)\n+\n+        //calculate P(c)\n+        // 1. Map: divide count of documents for a class through total count of documents\n+        //    (only element in documentCount data set)\n+        pc = documentsPerClass.map(new RichMapFunction[(String, Int), (String, Double)] {\n+\n+            var broadcastSet: util.List[Double] = null\n+\n+            override def open(config: Configuration): Unit = {\n+              broadcastSet = getRuntimeContext.getBroadcastVariable[Double](\"documentCount\")\n+              if (broadcastSet.size() != 1) {\n+                throw new RuntimeException(\"The document count data set used by p1 = 1 has the \" +\n+                  \"wrong size! Please use p1 = 0 if the problem can not be solved.\")\n+              }\n+            }\n+\n+            override def map(value: (String, Int)): (String, Double) = {\n+              (value._1, value._2 / broadcastSet.get(0))\n+            }\n+          }).withBroadcastSet(documentCount, \"documentCount\")\n+      }\n+      //END POSSIBILITY 1\n+\n+      // (list of all words, but distinct)\n+      val vocabulary = singleWordsInClass.map(tuple => (tuple._2, 1)).distinct(0)\n+      // (count of items in vocabulary list)\n+      val vocabularyCount: Double = vocabulary.count()\n+\n+      //calculate the P(w|c) value for words, that are not part of a class, needed for smoothing\n+      // 1. Map: use P(w|c) formula with smoothing with n(c_j, w_t) = 0\n+      val pwcNotInClass: DataSet[(String, Double)] = allWordsInClass\n+        .map(line =>\n+          (line._1, 1 / (line._2 + vocabularyCount))) // (class name -> P(w|c) word not in class)\n+\n+      //SCHNEIDER/RENNIE 1: ignore/reduce word frequency information\n+        //The singleWordsInClass data set must be changed before, the calculation of pwc starts for\n+        //schneider, it needs this form classname -> word -> number of documents containing wt in cj\n+\n+      if (sr1 == 1) {\n+        //Calculate the required data set (see above)\n+        // 1. FlatMap: class -> word -> 1 (one tuple for each document in which this word occurs)\n+        // 2. Group-Reduce: sum all 1s where the first two fields equal\n+        // 3. Map: Remove unesseccary count of word and replace with 1\n+        singleWordsInClass = input\n+          .flatMap(new SingleDistinctWordSplitter())\n+          .groupBy(0, 1)\n+          .reduce((line1, line2) => (line1._1, line1._2, line1._3 + line2._3))\n+      }\n+\n+      //END SCHNEIDER/RENNIE 1\n+\n+      //POSSIBILITY 3: way of calculating pwc\n+\n+      val p3 = resultingParameters(P3)\n+\n+      var pwc: DataSet[(String, String, Double)] = null // (class name -> word -> P(w|c))\n+\n+      if (p3 == 0) {\n+\n+          //Join the singleWordsInClass data set with the allWordsInClass data set to use the\n+          //information for the calculation of p(w|c).\n+          val wordsInClass = singleWordsInClass\n+            .join(allWordsInClass).where(0).equalTo(0) {\n+            (single, all) => (single._1, single._2, single._3, all._2)\n+          } // (class name -> word -> count of that word -> count of all words in that class)\n+\n+          //calculate the P(w|c) value for each word in each class\n+          // 1. Map: use normal P(w|c) formula\n+          pwc = wordsInClass.map(line => (line._1, line._2, (line._3 + 1) /\n+            (line._4 + vocabularyCount)))\n+\n+      } else if (p3 == 1) {\n+\n+        //calculate the P(w|c) value for each word in class\n+        //  1. Map: use normal P(w|c) formula / use the\n+        pwc = singleWordsInClass.map(new RichMapFunction[(String, String, Int),\n+          (String, String, Double)] {\n+\n+          var broadcastMap: mutable.Map[String, Int] = mutable.Map[String, Int]()\n+\n+\n+          override def open(config: Configuration): Unit = {\n+            val collection = getRuntimeContext\n+              .getBroadcastVariable[(String, Int)](\"allWordsInClass\").asScala\n+            for (record <- collection) {\n+              broadcastMap.put(record._1, record._2)\n+            }\n+          }\n+\n+          override def map(value: (String, String, Int)): (String, String, Double) = {\n+            (value._1, value._2, (value._3 + 1) / (broadcastMap(value._1) + vocabularyCount))\n+          }\n+        }).withBroadcastSet(allWordsInClass, \"allWordsInClass\")\n+\n+      }\n+\n+      //END POSSIBILITY 3\n+\n+      //stores all the word related information in one data set\n+      // 1. Map: Caluclate logarithms\n+      val wordRelatedModelData = pwc.map(line => (line._1, line._2, Math.log(line._3)))\n+\n+      //store all class related information in one data set\n+      // 1. Join: P(c) data set and P(w|c) data set not in class and calculate logarithms\n+      val classRelatedModelData = pc.join(pwcNotInClass)\n+        .where(0).equalTo(0) {\n+        (line1, line2) => (line1._1, Math.log(line1._2), Math.log(line2._2))\n+      } // (class name -> log(P(c)) -> log(P(w|c) not in class))\n+\n+      instance.wordRelatedModelData = Some(wordRelatedModelData)\n+      instance.classRelatedModelData = Some(classRelatedModelData)\n+\n+      //RENNIE 1: transform document frequency\n+        //for this, the improvementData set must be set\n+        //calculate (word -> log number of documents in all classes / docs with that word)\n+\n+      val r1 = resultingParameters(R1)\n+\n+      if (r1 == 1) {\n+        val totalDocumentCount: DataSet[(Double)] = documentsPerClass\n+          .reduce((line1, line2) => (line1._1, line1._2 + line2._2))\n+          .map(line => line._2) // (count of all documents)\n+\n+        //number of occurences over all documents of all classes\n+        val wordCountTotal = input\n+          .flatMap(new SingleDistinctWordSplitter())\n+          .map(line => (line._2, 1))\n+          .groupBy(0)\n+          .reduce((line1, line2) => (line1._1, line1._2 + line2._2))\n+           // (word -> count of documents with that word)\n+\n+        val improvementData = wordCountTotal.map(new RichMapFunction[(String, Int),\n+          (String, Double)] {\n+\n+          var broadcastSet: util.List[Double] = null\n+\n+          override def open(config: Configuration): Unit = {\n+            broadcastSet = getRuntimeContext.getBroadcastVariable[Double](\"totalDocumentCount\")\n+            if (broadcastSet.size() != 1) {\n+              throw new RuntimeException(\"The total document count data set used by 11 = 1 has \" +\n+                \"the wrong size! Please use r1 = 0 if the problem can not be solved.\")\n+            }\n+          }\n+\n+          override def map(value: (String, Int)): (String, Double) = {\n+            (value._1, Math.log(broadcastSet.get(0) / value._2))\n+          }\n+        }).withBroadcastSet(totalDocumentCount, \"totalDocumentCount\")\n+\n+        instance.improvementData = Some(improvementData)\n+      }\n+\n+    }\n+  }\n+\n+  // Model (String, String, Double, Double, Double)\n+  implicit def predictNNB = new PredictDataSetOperation[\n+    MultinomialNaiveBayes,\n+    (Int, String),\n+    (Int, String)]() {\n+\n+    override def predictDataSet(instance: MultinomialNaiveBayes,\n+                                predictParameters: ParameterMap,\n+                                input: DataSet[(Int, String)]): DataSet[(Int, String)] = {\n+\n+      if (instance.wordRelatedModelData.isEmpty || instance.classRelatedModelData.isEmpty) {\n+        throw new RuntimeException(\"The NormalNaiveBayes has not been fitted to the \" +\n+            \"data. This is necessary before a prediction on other data can be made.\")\n+      }\n+\n+      val wordRelatedModelData = instance.wordRelatedModelData.get\n+      val classRelatedModelData = instance.classRelatedModelData.get\n+\n+      val resultingParameters = instance.parameters ++ predictParameters\n+\n+      //split the texts from the input data set into its words\n+      val words: DataSet[(Int, String)] = input.flatMap {\n+        pair => pair._2.split(\" \").map { word => (pair._1, word)}\n+      }\n+\n+      //genreate word counts for each word with a key\n+      // 1. Map: put a 1 to each key\n+      // 2. Group-Reduce: group by id and word and sum the 1s\n+      val wordsAndCount: DataSet[(Int, String, Int)] = words.map(line => (line._1, line._2, 1))\n+        .groupBy(0, 1).sum(2) // (id -> word -> word count in text)\n+\n+      //calculate the count of all words for a text identified by its key\n+      val wordsInText: DataSet[(Int, Int)] = wordsAndCount.map(line => (line._1, line._3))\n+        .groupBy(0).sum(1) //(id -> all words in text)\n+\n+      //generate a data set containing all words that are in model for each id, class pair\n+      // 1. Join: wordRelatedModelData with wordsAndCount on\n+      //  words (id -> class -> word ->  word count -> log(P(w|c))\n+      val foundWords: DataSet[(Int, String, String, Int, Double)] = wordRelatedModelData\n+        .joinWithHuge(wordsAndCount).where(1).equalTo(1) {\n+        (wordR, wordsAC) => (wordsAC._1, wordR._1, wordsAC._2, wordsAC._3, wordR._3)\n+      }\n+\n+      //SCHNEIDER/RENNIE 1: ignore/reduce word frequency information\n+      //RENNIE 1: transform document frequency\n+\n+      val sr1 = resultingParameters(SR1)\n+      val r1 = resultingParameters(R1)\n+      var improvementData: DataSet[(String, Double)] = null\n+\n+      if (r1 == 1) {\n+        //The improvementData data set is needed\n+        if (instance.improvementData.isEmpty) {\n+          throw new RuntimeException(\"R1 = 1, for that additional data is needed, but it was not\" +\n+            \"found. Make sure to set R1 = 1 when fitting the training data.\")\n+        }\n+        improvementData = instance.improvementData.get\n+      }\n+\n+      if (sr1 == 1 && r1 == 1) {\n+        throw new RuntimeException(\"Parameter sr1 and r1 are both set to 1, which is not allowed.\")\n+      }\n+\n+      var sumPwcFoundWords: DataSet[(Int, String, Double)] = null\n+\n+      if (sr1 == 0 && r1 == 0) {\n+        //calculate sumpwc for found words\n+        // 1. Map: Remove unneded information from foundWords and calculate the sumP(w|c) for each\n+        //  word (id -> class -> word count * log(P(w|c))\n+        // 2. Group-Reduce: on id and class, sum all (word count * log(P(w|c))) results\n+        sumPwcFoundWords = foundWords\n+          .map(line => (line._1, line._2, line._4 * line._5))\n+          .groupBy(0, 1).reduce((line1, line2) =>\n+          (line1._1, line1._2, line1._3 + line2._3)) //(id -> class -> sum(log(P(w|c))\n+      } else if (sr1 == 1 && r1 == 0) {\n+        //same as sr1 == 0, but there is no multiplication with the word counts\n+        sumPwcFoundWords = foundWords\n+          .map(line => (line._1, line._2, line._5))\n+          .groupBy(0, 1).reduce((line1, line2) =>\n+          (line1._1, line1._2, line1._3 + line2._3)) //(id -> class -> sum(log(P(w|c))\n+      } else if (sr1 == 2 && r1 == 0) {\n+        //same es sr1 == 0, but multiplication with log(wordcount + 1)\n+        sumPwcFoundWords = foundWords\n+          .map(line => (line._1, line._2, Math.log(line._4 + 1) * line._5))\n+          .groupBy(0, 1).reduce((line1, line2) =>\n+          (line1._1, line1._2, line1._3 + line2._3)) //(id -> class -> sum(log(P(w|c))\n+      } else if (sr1 == 0 && r1 == 1) {\n+        //same as r1 = 0, but the word frequency is multiplied with with log (n_d(c) / n_d(w_t))\n+        //for that a join with the improvementData data set must be performed first to get the\n+        //needed additional information.\n+        // Join: (id -> class ->  word count * log P(w|c) * log (number of documents in class /\n+                                                            // word frequency in all classes)\n+        sumPwcFoundWords = foundWords\n+          .joinWithTiny(improvementData).where(2).equalTo(0) {\n+          (found, imp) => (found._1, found._2, found._4 * found._5 * imp._2)\n+          }.groupBy(0, 1).reduce((line1, line2) =>\n+          (line1._1, line1._2, line1._3 + line2._3)) //(id -> class -> sum(log(P(w|c)) */\n+      } else if (sr1 == 2 && r1 == 1) {\n+        //combination of r1 = 1 and sr1 =2\n+        sumPwcFoundWords = foundWords\n+          .joinWithTiny(improvementData).where(2).equalTo(0) {\n+          (found, imp) => (found._1, found._2, Math.log(found._4 + 1) * found._5 * imp._2)\n+        }.groupBy(0, 1).reduce((line1, line2) =>\n+          (line1._1, line1._2, line1._3 + line2._3)) //(id -> class -> sum(log(P(w|c)) */\n+      }\n+\n+      var sumPwcNotFoundWords: DataSet[(Int, String, Double)] = null\n+\n+      if (sr1 == 0 && r1 == 0) {\n+        //calculate sumwpc for words that are not in model in that class\n+        // 1. Map: Discard word and log(P(w|c) from foundWords\n+        // 2. Group-Reduce: calculate sum count of found words for each document,\n+        //  class pair (id -> class -> sum(wordCount))\n+        // 3. Join: with wordsInText on id, to get the count of all words per document\n+        // 4. Map: calculate sumPWcNotFound (id -> class ->\n+        //  (all words in document - found word in document) *\n+        //  log(P(w|c) not in class (provided by broadcast))\n+        sumPwcNotFoundWords = foundWords\n+          .map(line => (line._1, line._2, line._4))\n+          .groupBy(0, 1)\n+          .reduce((line1, line2) => (line1._1, line1._2, line1._3 + line2._3))\n+          .join(wordsInText).where(0).equalTo(0) {\n+          (foundW, wordsIT) => (foundW._1, foundW._2, foundW._3, wordsIT._2)\n+        }.map(new RichMapFunction[(Int, String, Int, Int), (Int, String, Double)] {\n+\n+          var broadcastMap: mutable.Map[String, Double] = mutable\n+            .Map[String, Double]() //class -> log(P(w|c) not found word in class)\n+\n+          override def open(config: Configuration): Unit = {\n+            val collection = getRuntimeContext\n+              .getBroadcastVariable[(String, Double, Double)](\"classRelatedModelData\")\n+              .asScala\n+            for (record <- collection) {\n+              broadcastMap.put(record._1, record._3)\n+            }\n+          }\n+\n+          override def map(value: (Int, String, Int, Int)): (Int, String, Double) = {\n+            (value._1, value._2, (value._4 - value._3) * broadcastMap(value._2))\n+          }\n+        }).withBroadcastSet(classRelatedModelData, \"classRelatedModelData\")\n+\n+      } else {\n+        /** if the word frequency is changed (as in SR1 = 1, SR1 = 2, R1 = 1), the\n+          * sumPwcNotFoundWords can not be calculated as above. They must be calculated the same way\n+          * as for the sumPwcFoundWords data set (because each word frequency for each word is\n+          * important).\n+          *\n+          * Prepare foundWords data set for differenz\n+          * 1. Map: discard log(P(w|c) and wordCount from foundWords (id -> class -> word)\n+          * = all words that are known in each document for each class\n+          *\n+          * Prepare wordsAndCounts data set for difference\n+          * 1. FlatMap:create for every tuple a (document id -> class -> word -> word count)\n+          *    tuple for every class\n+          *\n+          * Create notFoundWords data set\n+          * 1: CoGroup: Only tuples (id, class, word) that are not in preparedFoundWords\n+          *\n+          * The result is a data set, that contains all words for each document for each class\n+          * that are not part of that class and the word counts.\n+          *\n+          * Then calcualte sumPwcNotfoundWords\n+          */\n+\n+        val preparedFoundWords: DataSet[(Int, String, String)] = foundWords\n+          .map(line => (line._1, line._2, line._3))\n+\n+        val wordsAndCountsExtended: DataSet[(Int, String, String, Int)] = wordsAndCount\n+          .flatMap(new Extender())\n+          .withBroadcastSet(classRelatedModelData, \"classRelatedModelData\")\n+\n+        val notFoundWords: DataSet[(Int, String, String, Int)] = wordsAndCountsExtended\n+          .coGroup(preparedFoundWords).where(0, 1, 2)\n+          .equalTo(0, 1, 2)(new DifferenceCoGrouper) //(id -> class -> word -> word count)\n+\n+        if ((sr1 == 1 && r1 == 0) || (sr1 == 2 && r1 == 0)) {\n+          //calculate the sum of all Pwc for every not found word (id -> class -> sumPwcNotFound)\n+          // 1. Map: calculate the pwc value for every word (id -> class -> pwc)\n+          // 2. Sum: Sum these pwc values for each class and document\n+          sumPwcNotFoundWords = notFoundWords\n+            .map(new RichMapFunction[(Int, String, String, Int), (Int, String, Double)] {\n+\n+            var broadcastMap: mutable.Map[String, Double] = mutable\n+              .Map[String, Double]() //class -> log(P(w|c) not found word in class)\n+\n+            override def open(config: Configuration): Unit = {\n+              val collection = getRuntimeContext\n+                .getBroadcastVariable[(String, Double, Double)](\"classRelatedModelData\")\n+                .asScala\n+              for (record <- collection) {\n+                broadcastMap.put(record._1, record._3)\n+              }\n+            }\n+\n+            override def map(value: (Int, String, String, Int)): (Int, String, Double) = {\n+              if (sr1 == 1 && r1 == 0) {\n+                //same as sr1 == 0, but there is no multiplication with the word counts\n+                return (value._1, value._2, broadcastMap(value._2))\n+              } else if (sr1 == 2 && r1 == 0) {\n+                //same es sr1 == 0, but multiplication with log(wordcount + 1)\n+                return (value._1, value._2, Math.log(value._4 + 1) * broadcastMap(value._2))\n+              }\n+              throw new RuntimeException(\"sumPwcNotFound could not be calculated because you\" +\n+                \"choosed a not allowed parameter combination.\")\n+            }\n+          }).withBroadcastSet(classRelatedModelData, \"classRelatedModelData\")\n+            .groupBy(0, 1).sum(2)\n+        } else if ((sr1 == 0 && r1 == 1) || (sr1 == 2 && r1 == 1)) {\n+\n+          //same as r1 = 0, but the word frequency is multiplied with\n+          //log (n_d(c) / n_d(w_t)) for that a join with the improvementData data set must be\n+          //performed first to get the needed additional information.\n+          // 1. Join with improvement data: (id -> class ->  word count ->\n+            // log (number of documents in class / word frequency in all classes))\n+          // 2a. Map: (id -> class -> word count * log (see above) * log(P(w|c))\n+          // 2b. Map: (id -> class -> log(word count + 1) * log (see above) * log(P(w|c))\n+          sumPwcNotFoundWords = notFoundWords\n+            .join(improvementData).where(2).equalTo(0) {\n+            (nf, imp) => (nf._1, nf._2, nf._4, imp._2)\n+          }.map(new RichMapFunction[(Int, String, Int, Double), (Int, String, Double)] {\n+\n+            var broadcastMap: mutable.Map[String, Double] = mutable\n+              .Map[String, Double]() //class -> log(P(w|c) not found word in class)\n+\n+            override def open(config: Configuration): Unit = {\n+              val collection = getRuntimeContext\n+                .getBroadcastVariable[(String, Double, Double)](\"classRelatedModelData\")\n+                .asScala\n+              for (record <- collection) {\n+                broadcastMap.put(record._1, record._3)\n+              }\n+            }\n+\n+            override def map(value: (Int, String, Int, Double)): (Int, String, Double) = {\n+              if (sr1 == 0 && r1 == 1) {\n+                return (value._1, value._2, value._3 * value._4 * broadcastMap(value._2))\n+              } else if (sr1 == 2 && r1 == 1) {\n+                return (value._1, value._2, Math.log(value._3 + 1)\n+                  * value._4 * broadcastMap(value._2))\n+              }\n+              throw new RuntimeException(\"sumPwcNotFound could not be calculated because you\" +\n+                \"choosed a not allowed parameter combination.\")\n+            }\n+          }).withBroadcastSet(classRelatedModelData, \"classRelatedModelData\")\n+            .groupBy(0, 1).sum(2) //(id -> class -> sum(log(P(w|c))\n+        }\n+      }\n+\n+      var sumPwc: DataSet[(Int, String, Double)] = null\n+\n+      if (sr1 == 0 && r1 == 0) {\n+        //sum those pwc sums\n+        // 1. Join sumPwcFoundWords and sumPwcNotFoundWords\n+        // 2. Map: add sums from found words and sums from not found words\n+        sumPwc = sumPwcFoundWords\n+          .join(sumPwcNotFoundWords).where(0, 1).equalTo(0, 1) {\n+          (found, notfound) => (found._1, found._2, found._3 + notfound._3)\n+        } //(id -> class name -> sum log(p(w|c)))\n+      } else {\n+        //sum those pwc sums\n+        // 1. CoGroup Found with NotFound on document id\n+        // add values together, when both have a same class\n+        // submit tuple, when a class misses for the other one\n+        sumPwc = sumPwcFoundWords.coGroup(sumPwcNotFoundWords)\n+          .where(0).equalTo(0) {\n+          (foundVs, notFoundVs, out: Collector[(Int, String, Double)]) =>\n+            //1. Create two Maps\n+            var foundMap = Map[String, Double]()\n+            var notFoundMap = Map[String, Double]()\n+\n+            var currentID = -1\n+\n+            for (foundV <- foundVs) {\n+              if (currentID == -1) {\n+                currentID = foundV._1\n+              }\n+              foundMap += foundV._2 -> foundV._3\n+            }\n+            for (notFoundV <- notFoundVs) {\n+              notFoundMap += notFoundV._2 -> notFoundV._3\n+            }\n+            //2. Go through these maps and collect tuples\n+            for (tuple <- foundMap) {\n+              if (notFoundMap.contains(tuple._1)) {\n+                out.collect(currentID, tuple._1, tuple._2 + notFoundMap(tuple._1))\n+                notFoundMap.remove(tuple._1)\n+              } else {\n+                out.collect(currentID, tuple._1, tuple._2)\n+              }\n+            }\n+            for (tuple <- notFoundMap) {\n+              if (foundMap.contains(tuple._1)) {\n+                out.collect(currentID, tuple._1, tuple._2 + foundMap(tuple._1))\n+              } else {\n+                out.collect(currentID, tuple._1, tuple._2)\n+              }\n+            }\n+\n+\n+        }\n+      }\n+\n+      //END RENNIE 1\n+      //END SCHNEIDER/RENNIE 1\n+\n+      //SCHNEIDER 1: ignore P(c_j) in cMAP formula\n+      val s1 = resultingParameters(S1)\n+      var posterior: DataSet[(Int, String, Double)] = null\n+      if (s1 == 0) {\n+        //calculate posterior values for each class\n+        // 1. Map: add sumPwc values with log(P(c)) (provided by broadcast)\n+        posterior = sumPwc\n+          .map(new RichMapFunction[(Int, String, Double), (Int, String, Double)] {\n+\n+          var broadcastMap: mutable.Map[String, Double] =\n+            mutable.Map[String, Double]() //class -> log(P(c))\n+\n+          override def open(config: Configuration): Unit = {\n+            val collection = getRuntimeContext\n+              .getBroadcastVariable[(String, Double, Double)](\"classRelatedModelData\")\n+              .asScala\n+            for (record <- collection) {\n+              broadcastMap.put(record._1, record._2)\n+            }\n+          }\n+\n+          override def map(value: (Int, String, Double)): (Int, String, Double) = {\n+            (value._1, value._2, value._3 + broadcastMap(value._2))\n+          }\n+        }).withBroadcastSet(classRelatedModelData, \"classRelatedModelData\")\n+      } else if (s1 == 1) {\n+        posterior = sumPwc\n+      }\n+      //choose the highest probable class\n+      // 1. Reduce: keep only highest probability\n+      posterior.groupBy(0).reduce(new CalculateReducer())\n+        .map(line => (line._1, line._2)) //(id -> class)\n+\n+\n+    }\n+  }\n+\n+\n+  /*\n+  * ************************************************************************************************\n+  * *******************************************Function Classes*************************************\n+  * ************************************************************************************************\n+   */\n+\n+  /**\n+   * Transforms a (String, String) tuple into a (String, String, Int)) tuple.\n+   * The second string from the input gets split into its words, for each word a tuple is collected\n+   * with the Int 1.\n+   */\n+  class SingleWordSplitter() extends FlatMapFunction[(String, String), (String, String, Int)] {\n+    override def flatMap(value: (String, String), out: Collector[(String, String, Int)]): Unit = {\n+      for (token: String <- value._2.split(\" \")) {\n+        out.collect((value._1, token, 1))\n+      }\n+    }\n+\n+    // Can be implemented via: input.flatMap{ pair => pair._2.split(\" \")\n+    // .map{ token => (pair._1, token ,1)} }\n+  }\n+\n+  /**\n+   * Transforms a (String, String) tuple into a (String, String, Int)) tuple.\n+   * The second string from the input gets split into its words, for each distinct word a tuple\n+   * is collected with the Int 1.\n+   */\n+  class SingleDistinctWordSplitter() extends FlatMapFunction[\n+    (String, String),\n+    (String, String, Int)] {\n+    override def flatMap(value: (String, String), out: Collector[(String, String, Int)]): Unit = {\n+      var x: ListBuffer[String] = ListBuffer()\n+      for (token: String <- value._2.split(\" \")) {\n+        x.append(token)\n+      }\n+      x = x.distinct\n+      for (item <- x) {\n+        out.collect((value._1, item, 1))\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Used to extend the wordsAndCount data set. Transforms (id -> word -> word count) into\n+   * (id -> class -> word -> word count) for each possible class.\n+   */\n+  class Extender() extends RichFlatMapFunction[(Int, String, Int), (Int, String, String, Int)] {\n+\n+    var broadcastSet = Set[String]()\n+\n+    override def open(config: Configuration): Unit = {\n+      val collection = getRuntimeContext\n+        .getBroadcastVariable[(String, Double, Double)](\"classRelatedModelData\")\n+        .asScala\n+      for (record <- collection) {\n+        broadcastSet += record._1\n+      }\n+    }\n+\n+    override def flatMap(value: (Int, String, Int), out: Collector[(Int, String, String, Int)]):\n+      Unit = {\n+      for (c: String <- broadcastSet) {\n+        out.collect((value._1, c, value._2, value._3))\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Chooses for each label that class with the highest value\n+   */\n+  class CalculateReducer() extends ReduceFunction[(Int, String, Double)] {\n+    override def reduce(value1: (Int, String, Double),","path":"flink-staging/flink-ml/src/main/scala/org/apache/flink/ml/classification/MultinomialNaiveBayes.scala","position":873,"original_position":873,"commit_id":"fadd761e175f2d7308e77f3acab03ce461b0abd5","original_commit_id":"fadd761e175f2d7308e77f3acab03ce461b0abd5","user":{"login":"sachingoel0101","id":8874261,"avatar_url":"https://avatars.githubusercontent.com/u/8874261?v=3","gravatar_id":"","url":"https://api.github.com/users/sachingoel0101","html_url":"https://github.com/sachingoel0101","followers_url":"https://api.github.com/users/sachingoel0101/followers","following_url":"https://api.github.com/users/sachingoel0101/following{/other_user}","gists_url":"https://api.github.com/users/sachingoel0101/gists{/gist_id}","starred_url":"https://api.github.com/users/sachingoel0101/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sachingoel0101/subscriptions","organizations_url":"https://api.github.com/users/sachingoel0101/orgs","repos_url":"https://api.github.com/users/sachingoel0101/repos","events_url":"https://api.github.com/users/sachingoel0101/events{/privacy}","received_events_url":"https://api.github.com/users/sachingoel0101/received_events","type":"User","site_admin":false},"body":"indentation","created_at":"2015-09-25T15:53:35Z","updated_at":"2015-09-25T15:53:35Z","html_url":"https://github.com/apache/flink/pull/1156#discussion_r40444866","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1156","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/40444866"},"html":{"href":"https://github.com/apache/flink/pull/1156#discussion_r40444866"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1156"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1156","id":45539137,"html_url":"https://github.com/apache/flink/pull/1156","diff_url":"https://github.com/apache/flink/pull/1156.diff","patch_url":"https://github.com/apache/flink/pull/1156.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1156","number":1156,"state":"open","locked":false,"title":"[FLINK-1719] [ml] Add Multinomial Naive Bayes Classification","user":{"login":"JonathanH5","id":5738978,"avatar_url":"https://avatars.githubusercontent.com/u/5738978?v=3","gravatar_id":"","url":"https://api.github.com/users/JonathanH5","html_url":"https://github.com/JonathanH5","followers_url":"https://api.github.com/users/JonathanH5/followers","following_url":"https://api.github.com/users/JonathanH5/following{/other_user}","gists_url":"https://api.github.com/users/JonathanH5/gists{/gist_id}","starred_url":"https://api.github.com/users/JonathanH5/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JonathanH5/subscriptions","organizations_url":"https://api.github.com/users/JonathanH5/orgs","repos_url":"https://api.github.com/users/JonathanH5/repos","events_url":"https://api.github.com/users/JonathanH5/events{/privacy}","received_events_url":"https://api.github.com/users/JonathanH5/received_events","type":"User","site_admin":false},"body":"This pull request is related to [FLINK-1719](https://issues.apache.org/jira/browse/FLINK-1719).\r\nMultinomial Naive Bayes was successfully implemented @tillrohrmann and different ideas proposed by other authors were incorporated.\r\n\r\n","created_at":"2015-09-21T11:47:18Z","updated_at":"2015-09-25T15:53:35Z","closed_at":null,"merged_at":null,"merge_commit_sha":"831f846a8d6ee29f6374eb6257978a932eee5e1f","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1156/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1156/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1156/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/fadd761e175f2d7308e77f3acab03ce461b0abd5","head":{"label":"JonathanH5:pullrequest","ref":"pullrequest","sha":"fadd761e175f2d7308e77f3acab03ce461b0abd5","user":{"login":"JonathanH5","id":5738978,"avatar_url":"https://avatars.githubusercontent.com/u/5738978?v=3","gravatar_id":"","url":"https://api.github.com/users/JonathanH5","html_url":"https://github.com/JonathanH5","followers_url":"https://api.github.com/users/JonathanH5/followers","following_url":"https://api.github.com/users/JonathanH5/following{/other_user}","gists_url":"https://api.github.com/users/JonathanH5/gists{/gist_id}","starred_url":"https://api.github.com/users/JonathanH5/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JonathanH5/subscriptions","organizations_url":"https://api.github.com/users/JonathanH5/orgs","repos_url":"https://api.github.com/users/JonathanH5/repos","events_url":"https://api.github.com/users/JonathanH5/events{/privacy}","received_events_url":"https://api.github.com/users/JonathanH5/received_events","type":"User","site_admin":false},"repo":{"id":34563457,"name":"flink","full_name":"JonathanH5/flink","owner":{"login":"JonathanH5","id":5738978,"avatar_url":"https://avatars.githubusercontent.com/u/5738978?v=3","gravatar_id":"","url":"https://api.github.com/users/JonathanH5","html_url":"https://github.com/JonathanH5","followers_url":"https://api.github.com/users/JonathanH5/followers","following_url":"https://api.github.com/users/JonathanH5/following{/other_user}","gists_url":"https://api.github.com/users/JonathanH5/gists{/gist_id}","starred_url":"https://api.github.com/users/JonathanH5/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JonathanH5/subscriptions","organizations_url":"https://api.github.com/users/JonathanH5/orgs","repos_url":"https://api.github.com/users/JonathanH5/repos","events_url":"https://api.github.com/users/JonathanH5/events{/privacy}","received_events_url":"https://api.github.com/users/JonathanH5/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/JonathanH5/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/JonathanH5/flink","forks_url":"https://api.github.com/repos/JonathanH5/flink/forks","keys_url":"https://api.github.com/repos/JonathanH5/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/JonathanH5/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/JonathanH5/flink/teams","hooks_url":"https://api.github.com/repos/JonathanH5/flink/hooks","issue_events_url":"https://api.github.com/repos/JonathanH5/flink/issues/events{/number}","events_url":"https://api.github.com/repos/JonathanH5/flink/events","assignees_url":"https://api.github.com/repos/JonathanH5/flink/assignees{/user}","branches_url":"https://api.github.com/repos/JonathanH5/flink/branches{/branch}","tags_url":"https://api.github.com/repos/JonathanH5/flink/tags","blobs_url":"https://api.github.com/repos/JonathanH5/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/JonathanH5/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/JonathanH5/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/JonathanH5/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/JonathanH5/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/JonathanH5/flink/languages","stargazers_url":"https://api.github.com/repos/JonathanH5/flink/stargazers","contributors_url":"https://api.github.com/repos/JonathanH5/flink/contributors","subscribers_url":"https://api.github.com/repos/JonathanH5/flink/subscribers","subscription_url":"https://api.github.com/repos/JonathanH5/flink/subscription","commits_url":"https://api.github.com/repos/JonathanH5/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/JonathanH5/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/JonathanH5/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/JonathanH5/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/JonathanH5/flink/contents/{+path}","compare_url":"https://api.github.com/repos/JonathanH5/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/JonathanH5/flink/merges","archive_url":"https://api.github.com/repos/JonathanH5/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/JonathanH5/flink/downloads","issues_url":"https://api.github.com/repos/JonathanH5/flink/issues{/number}","pulls_url":"https://api.github.com/repos/JonathanH5/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/JonathanH5/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/JonathanH5/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/JonathanH5/flink/labels{/name}","releases_url":"https://api.github.com/repos/JonathanH5/flink/releases{/id}","created_at":"2015-04-25T09:53:54Z","updated_at":"2015-04-25T09:54:04Z","pushed_at":"2015-09-22T11:16:00Z","git_url":"git://github.com/JonathanH5/flink.git","ssh_url":"git@github.com:JonathanH5/flink.git","clone_url":"https://github.com/JonathanH5/flink.git","svn_url":"https://github.com/JonathanH5/flink","homepage":null,"size":61447,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"64e1dc6fe2c4970460c286a36136e9c4c32540a8","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","created_at":"2014-06-07T07:00:10Z","updated_at":"2015-09-25T03:05:24Z","pushed_at":"2015-09-25T15:27:48Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":2524860,"stargazers_count":638,"watchers_count":638,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":430,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":74,"forks":430,"open_issues":74,"watchers":638,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1156"},"html":{"href":"https://github.com/apache/flink/pull/1156"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1156"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1156/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1156/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1156/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/fadd761e175f2d7308e77f3acab03ce461b0abd5"}}}}