You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sdap.apache.org by le...@apache.org on 2017/12/19 14:13:07 UTC
[06/17] incubator-sdap-mudrod git commit: SDAP-7 Change all package
namespaces to org.apache.sdap
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/utils/MatrixUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/utils/MatrixUtil.java b/core/src/main/java/org/apache/sdap/mudrod/utils/MatrixUtil.java
new file mode 100644
index 0000000..8259ce7
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/utils/MatrixUtil.java
@@ -0,0 +1,488 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.utils;
+
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.Optional;
+import org.apache.spark.api.java.function.*;
+import org.apache.spark.mllib.feature.IDF;
+import org.apache.spark.mllib.feature.IDFModel;
+import org.apache.spark.mllib.linalg.*;
+import org.apache.spark.mllib.linalg.distributed.IndexedRow;
+import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;
+import org.apache.spark.mllib.linalg.distributed.RowMatrix;
+import scala.Tuple2;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Stream;
+
+/**
+ * Matrix utility tool
+ */
+public class MatrixUtil {
+
+ private MatrixUtil() {
+ }
+
+ /**
+ * buildSVDMatrix: Generate SVD matrix from TF-IDF matrix. Please make sure
+ * the TF-IDF matrix has been already built from the original documents.
+ *
+ * @param tfidfMatrix,
+ * each row is a term and each column is a document name and each
+ * cell is the TF-IDF value of the term in the corresponding
+ * document.
+ * @param dimension
+ * Column number of the SVD matrix
+ * @return RowMatrix, each row is a term and each column is a dimension in the
+ * feature space, each cell is value of the term in the corresponding
+ * dimension.
+ */
+ public static RowMatrix buildSVDMatrix(RowMatrix tfidfMatrix, int dimension) {
+ int matrixCol = (int) tfidfMatrix.numCols();
+ if (matrixCol < dimension) {
+ dimension = matrixCol;
+ }
+
+ SingularValueDecomposition<RowMatrix, Matrix> svd = tfidfMatrix.computeSVD(dimension, true, 1.0E-9d);
+ RowMatrix u = svd.U();
+ Vector s = svd.s();
+ return u.multiply(Matrices.diag(s));
+ }
+
+ /**
+ * buildSVDMatrix: Generate SVD matrix from Vector RDD.
+ *
+ * @param vecRDD
+ * vectors of terms in feature space
+ * @param dimension
+ * Column number of the SVD matrix
+ * @return RowMatrix, each row is a term and each column is a dimension in the
+ * feature space, each cell is value of the term in the corresponding
+ * dimension.
+ */
+ public static RowMatrix buildSVDMatrix(JavaRDD<Vector> vecRDD, int dimension) {
+ RowMatrix tfidfMatrix = new RowMatrix(vecRDD.rdd());
+ SingularValueDecomposition<RowMatrix, Matrix> svd = tfidfMatrix.computeSVD(dimension, true, 1.0E-9d);
+ RowMatrix u = svd.U();
+ Vector s = svd.s();
+ return u.multiply(Matrices.diag(s));
+ }
+
+ /**
+ * Create TF-IDF matrix from word-doc matrix.
+ *
+ * @param wordDocMatrix,
+ * each row is a term, each column is a document name and each cell
+ * is number of the term in the corresponding document.
+ * @return RowMatrix, each row is a term and each column is a document name
+ * and each cell is the TF-IDF value of the term in the corresponding
+ * document.
+ */
+ public static RowMatrix createTFIDFMatrix(RowMatrix wordDocMatrix) {
+ JavaRDD<Vector> newcountRDD = wordDocMatrix.rows().toJavaRDD();
+ IDFModel idfModel = new IDF().fit(newcountRDD);
+ JavaRDD<Vector> idf = idfModel.transform(newcountRDD);
+ return new RowMatrix(idf.rdd());
+ }
+
+ /**
+ * Create matrix from doc-terms JavaPairRDD.
+ *
+ * @param uniqueDocRDD
+ * doc-terms JavaPairRDD, in which each key is a doc name, and value
+ * is term list extracted from that doc
+ * @return LabeledRowMatrix {@link LabeledRowMatrix}
+ */
+ public static LabeledRowMatrix createWordDocMatrix(JavaPairRDD<String, List<String>> uniqueDocRDD) {
+ // Index documents with unique IDs
+ JavaPairRDD<List<String>, Long> corpus = uniqueDocRDD.values().zipWithIndex();
+ // cal word-doc numbers
+ JavaPairRDD<Tuple2<String, Long>, Double> worddocNumRDD = corpus.flatMapToPair(new PairFlatMapFunction<Tuple2<List<String>, Long>, Tuple2<String, Long>, Double>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Iterator<Tuple2<Tuple2<String, Long>, Double>> call(Tuple2<List<String>, Long> docwords) throws Exception {
+ List<Tuple2<Tuple2<String, Long>, Double>> pairs = new ArrayList<>();
+ List<String> words = docwords._1;
+ int n = words.size();
+ for (int i = 0; i < n; i++) {
+ Tuple2<String, Long> worddoc = new Tuple2<>(words.get(i), docwords._2);
+ pairs.add(new Tuple2<Tuple2<String, Long>, Double>(worddoc, 1.0));
+ }
+ return pairs.iterator();
+ }
+ }).reduceByKey(new Function2<Double, Double, Double>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Double call(Double first, Double second) throws Exception {
+ return first + second;
+ }
+ });
+ // cal word doc-numbers
+ JavaPairRDD<String, Tuple2<List<Long>, List<Double>>> wordDocnumRDD = worddocNumRDD
+ .mapToPair(new PairFunction<Tuple2<Tuple2<String, Long>, Double>, String, Tuple2<List<Long>, List<Double>>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Tuple2<List<Long>, List<Double>>> call(Tuple2<Tuple2<String, Long>, Double> worddocNum) throws Exception {
+ List<Long> docs = new ArrayList<>();
+ docs.add(worddocNum._1._2);
+ List<Double> nums = new ArrayList<>();
+ nums.add(worddocNum._2);
+ Tuple2<List<Long>, List<Double>> docmums = new Tuple2<>(docs, nums);
+ return new Tuple2<>(worddocNum._1._1, docmums);
+ }
+ });
+ // trans to vector
+ final int corporsize = (int) uniqueDocRDD.keys().count();
+ JavaPairRDD<String, Vector> wordVectorRDD = wordDocnumRDD.reduceByKey(new Function2<Tuple2<List<Long>, List<Double>>, Tuple2<List<Long>, List<Double>>, Tuple2<List<Long>, List<Double>>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<List<Long>, List<Double>> call(Tuple2<List<Long>, List<Double>> arg0, Tuple2<List<Long>, List<Double>> arg1) throws Exception {
+ arg0._1.addAll(arg1._1);
+ arg0._2.addAll(arg1._2);
+ return new Tuple2<>(arg0._1, arg0._2);
+ }
+ }).mapToPair(new PairFunction<Tuple2<String, Tuple2<List<Long>, List<Double>>>, String, Vector>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Vector> call(Tuple2<String, Tuple2<List<Long>, List<Double>>> arg0) throws Exception {
+ int docsize = arg0._2._1.size();
+ int[] intArray = new int[docsize];
+ double[] doubleArray = new double[docsize];
+ for (int i = 0; i < docsize; i++) {
+ intArray[i] = arg0._2._1.get(i).intValue();
+ doubleArray[i] = arg0._2._2.get(i).intValue();
+ }
+ Vector sv = Vectors.sparse(corporsize, intArray, doubleArray);
+ return new Tuple2<>(arg0._1, sv);
+ }
+ });
+
+ RowMatrix wordDocMatrix = new RowMatrix(wordVectorRDD.values().rdd());
+
+ LabeledRowMatrix labeledRowMatrix = new LabeledRowMatrix();
+ labeledRowMatrix.rowMatrix = wordDocMatrix;
+ labeledRowMatrix.rowkeys = wordVectorRDD.keys().collect();
+ labeledRowMatrix.colkeys = uniqueDocRDD.keys().collect();
+ return labeledRowMatrix;
+ }
+
+ public static LabeledRowMatrix createDocWordMatrix(JavaPairRDD<String, List<String>> uniqueDocRDD, JavaSparkContext sc) {
+ // Index word with unique IDs
+ JavaPairRDD<String, Long> wordIDRDD = uniqueDocRDD.values().flatMap(new FlatMapFunction<List<String>, String>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Iterator<String> call(List<String> arg0) throws Exception {
+ return arg0.iterator();
+ }
+ }).distinct().zipWithIndex();
+
+ //
+ JavaPairRDD<Tuple2<String, String>, Double> docwordNumRDD = uniqueDocRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, List<String>>, Tuple2<String, String>, Double>() {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Iterator<Tuple2<Tuple2<String, String>, Double>> call(Tuple2<String, List<String>> docwords) throws Exception {
+ List<Tuple2<Tuple2<String, String>, Double>> pairs = new ArrayList<>();
+ List<String> words = docwords._2;
+ int n = words.size();
+ for (int i = 0; i < n; i++) {
+ Tuple2<String, String> worddoc = new Tuple2<>(docwords._1, words.get(i));
+ pairs.add(new Tuple2<Tuple2<String, String>, Double>(worddoc, 1.0));
+ }
+ return pairs.iterator();
+ }
+ }).reduceByKey(new Function2<Double, Double, Double>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Double call(Double first, Double second) throws Exception {
+ return first + second;
+ }
+ });
+
+ //
+ JavaPairRDD<String, Tuple2<String, Double>> wordDocnumRDD = docwordNumRDD.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Double>, String, Tuple2<String, Double>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Tuple2<String, Double>> call(Tuple2<Tuple2<String, String>, Double> arg0) throws Exception {
+
+ Tuple2<String, Double> wordmums = new Tuple2<>(arg0._1._1, arg0._2);
+ return new Tuple2<>(arg0._1._2, wordmums);
+ }
+ });
+
+ //
+
+ JavaPairRDD<String, Tuple2<Tuple2<String, Double>, Optional<Long>>> testRDD = wordDocnumRDD.leftOuterJoin(wordIDRDD);
+
+ int wordsize = (int) wordIDRDD.count();
+ JavaPairRDD<String, Vector> docVectorRDD = testRDD.mapToPair(new PairFunction<Tuple2<String, Tuple2<Tuple2<String, Double>, Optional<Long>>>, String, Tuple2<List<Long>, List<Double>>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Tuple2<List<Long>, List<Double>>> call(Tuple2<String, Tuple2<Tuple2<String, Double>, Optional<Long>>> arg0) throws Exception {
+ Optional<Long> oid = arg0._2._2;
+ Long wordId = (long) 0;
+ if (oid.isPresent()) {
+ wordId = oid.get();
+ }
+
+ List<Long> word = new ArrayList<>();
+ word.add(wordId);
+
+ List<Double> count = new ArrayList<>();
+ count.add(arg0._2._1._2);
+
+ Tuple2<List<Long>, List<Double>> wordcount = new Tuple2<>(word, count);
+
+ return new Tuple2<>(arg0._2._1._1, wordcount);
+ }
+
+ }).reduceByKey(new Function2<Tuple2<List<Long>, List<Double>>, Tuple2<List<Long>, List<Double>>, Tuple2<List<Long>, List<Double>>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<List<Long>, List<Double>> call(Tuple2<List<Long>, List<Double>> arg0, Tuple2<List<Long>, List<Double>> arg1) throws Exception {
+ arg0._1.addAll(arg1._1);
+ arg0._2.addAll(arg1._2);
+ return new Tuple2<>(arg0._1, arg0._2);
+ }
+ }).mapToPair(new PairFunction<Tuple2<String, Tuple2<List<Long>, List<Double>>>, String, Vector>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Vector> call(Tuple2<String, Tuple2<List<Long>, List<Double>>> arg0) throws Exception {
+ int docsize = arg0._2._1.size();
+ int[] intArray = new int[docsize];
+ double[] doubleArray = new double[docsize];
+ for (int i = 0; i < docsize; i++) {
+ intArray[i] = arg0._2._1.get(i).intValue();
+ doubleArray[i] = arg0._2._2.get(i).intValue();
+ }
+ Vector sv = Vectors.sparse(wordsize, intArray, doubleArray);
+ return new Tuple2<>(arg0._1, sv);
+ }
+ });
+
+ RowMatrix docwordMatrix = new RowMatrix(docVectorRDD.values().rdd());
+
+ LabeledRowMatrix labeledRowMatrix = new LabeledRowMatrix();
+ labeledRowMatrix.rowMatrix = docwordMatrix;
+ labeledRowMatrix.rowkeys = docVectorRDD.keys().collect();
+ labeledRowMatrix.colkeys = wordIDRDD.keys().collect();
+
+ return labeledRowMatrix;
+ }
+
+ /**
+ * loadVectorFromCSV: Load term vector from csv file.
+ *
+ * @param spark
+ * spark instance
+ * @param csvFileName
+ * csv matrix file
+ * @param skipNum
+ * the numbers of rows which should be skipped Ignore the top skip
+ * number rows of the csv file
+ * @return JavaPairRDD, each key is a term, and value is the vector of the
+ * term in feature space.
+ */
+ public static JavaPairRDD<String, Vector> loadVectorFromCSV(SparkDriver spark, String csvFileName, int skipNum) {
+ // skip the first line (header), important!
+ JavaRDD<String> importRDD = spark.sc.textFile(csvFileName);
+ JavaPairRDD<String, Long> importIdRDD = importRDD.zipWithIndex().filter(new Function<Tuple2<String, Long>, Boolean>() {
+ /** */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Boolean call(Tuple2<String, Long> v1) throws Exception {
+ if (v1._2 > (skipNum - 1)) {
+ return true;
+ }
+ return false;
+ }
+ });
+
+ if (importIdRDD.count() == 0) {
+ return null;
+ }
+
+ return importIdRDD.mapToPair(new PairFunction<Tuple2<String, Long>, String, Vector>() {
+ /** */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<String, Vector> call(Tuple2<String, Long> t) throws Exception {
+ String[] fields = t._1.split(",");
+ String word = fields[0];
+ int fieldsize = fields.length;
+ int nStart = 1;
+ int nEnd = fieldsize - 1;
+ if (fieldsize < 2) {
+ nStart = 0;
+ nEnd = 0;
+ }
+ String[] numfields = Arrays.copyOfRange(fields, nStart, nEnd);
+
+ double[] nums = Stream.of(numfields).mapToDouble(Double::parseDouble).toArray();
+ Vector vec = Vectors.dense(nums);
+ return new Tuple2<>(word, vec);
+ }
+ });
+ }
+
+ /**
+ * Convert vectorRDD to indexed row matrix.
+ *
+ * @param vecs
+ * Vector RDD
+ * @return IndexedRowMatrix
+ */
+ public static IndexedRowMatrix buildIndexRowMatrix(JavaRDD<Vector> vecs) {
+ JavaRDD<IndexedRow> indexrows = vecs.zipWithIndex().map(new Function<Tuple2<Vector, Long>, IndexedRow>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public IndexedRow call(Tuple2<Vector, Long> docId) {
+ return new IndexedRow(docId._2, docId._1);
+ }
+ });
+ return new IndexedRowMatrix(indexrows.rdd());
+ }
+
+ /**
+ * Transpose matrix
+ *
+ * @param indexedMatrix
+ * spark indexed matrix
+ * @return rowmatrix, each row is corresponding to the column in the original
+ * matrix and vice versa
+ */
+ public static RowMatrix transposeMatrix(IndexedRowMatrix indexedMatrix) {
+ return indexedMatrix.toCoordinateMatrix().transpose().toRowMatrix();
+ }
+
+ /**
+ * Output matrix to a CSV file.
+ *
+ * @param matrix
+ * spark row matrix
+ * @param rowKeys
+ * matrix row names
+ * @param colKeys
+ * matrix coloum names
+ * @param fileName
+ * csv file name
+ */
+ public static void exportToCSV(RowMatrix matrix, List<String> rowKeys, List<String> colKeys, String fileName) {
+
+ if (matrix.rows().isEmpty()) {
+ return;
+ }
+
+ int rownum = (int) matrix.numRows();
+ int colnum = (int) matrix.numCols();
+ List<Vector> rows = matrix.rows().toJavaRDD().collect();
+
+ File file = new File(fileName);
+ if (file.exists()) {
+ file.delete();
+ }
+ try {
+ file.createNewFile();
+ FileWriter fw = new FileWriter(file.getAbsoluteFile());
+ BufferedWriter bw = new BufferedWriter(fw);
+ String coltitle = " Num" + ",";
+ for (int j = 0; j < colnum; j++) {
+ coltitle += "\"" + colKeys.get(j) + "\",";
+ }
+ coltitle = coltitle.substring(0, coltitle.length() - 1);
+ bw.write(coltitle + "\n");
+
+ for (int i = 0; i < rownum; i++) {
+ double[] rowvlaue = rows.get(i).toArray();
+ String row = rowKeys.get(i) + ",";
+ for (int j = 0; j < colnum; j++) {
+ row += rowvlaue[j] + ",";
+ }
+ row = row.substring(0, row.length() - 1);
+ bw.write(row + "\n");
+ }
+
+ bw.close();
+
+ } catch (IOException e) {
+ e.printStackTrace();
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/utils/RDDUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/utils/RDDUtil.java b/core/src/main/java/org/apache/sdap/mudrod/utils/RDDUtil.java
new file mode 100644
index 0000000..8c2e64c
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/utils/RDDUtil.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.utils;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.FlatMapFunction;
+
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * ClassName: RDDUtil Function: Mudrod Spark RDD common methods
+ */
+public class RDDUtil {
+
+ public RDDUtil() {
+ }
+
+ /**
+ * getAllWordsInDoc: Extracted all unique terms from all docs.
+ *
+ * @param docwordRDD Pair RDD, each key is a doc, and value is term list extracted from
+ * that doc.
+ * @return unique term list
+ */
+ public static JavaRDD<String> getAllWordsInDoc(JavaPairRDD<String, List<String>> docwordRDD) {
+ JavaRDD<String> wordRDD = docwordRDD.values().flatMap(new FlatMapFunction<List<String>, String>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Iterator<String> call(List<String> list) {
+ return list.iterator();
+ }
+ }).distinct();
+
+ return wordRDD;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/utils/SVDUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/utils/SVDUtil.java b/core/src/main/java/org/apache/sdap/mudrod/utils/SVDUtil.java
new file mode 100644
index 0000000..1cd4e00
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/utils/SVDUtil.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.utils;
+
+import org.apache.sdap.mudrod.discoveryengine.MudrodAbstract;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.mllib.linalg.Vector;
+import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix;
+import org.apache.spark.mllib.linalg.distributed.RowMatrix;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Singular value decomposition
+ */
+public class SVDUtil extends MudrodAbstract {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ // wordRDD: terms extracted from all documents
+ JavaRDD<String> wordRDD;
+ // svdMatrix: svd matrix
+ private RowMatrix svdMatrix;
+ // simMatrix: similarity matrix
+ private CoordinateMatrix simMatrix;
+
+ /**
+ * Creates a new instance of SVDUtil.
+ *
+ * @param config the Mudrod configuration
+ * @param es the Elasticsearch drive
+ * @param spark the spark driver
+ */
+ public SVDUtil(Properties config, ESDriver es, SparkDriver spark) {
+ super(config, es, spark);
+ }
+
+ /**
+ * Build SVD matrix from docment-terms pairs.
+ *
+ * @param docwordRDD JavaPairRDD, key is short name of data set and values are terms in
+ * the corresponding data set
+ * @param svdDimension: Dimension of matrix after singular value decomposition
+ * @return row matrix
+ */
+ public RowMatrix buildSVDMatrix(JavaPairRDD<String, List<String>> docwordRDD, int svdDimension) {
+
+ RowMatrix svdMatrix = null;
+ LabeledRowMatrix wordDocMatrix = MatrixUtil.createWordDocMatrix(docwordRDD);
+ RowMatrix ifIdfMatrix = MatrixUtil.createTFIDFMatrix(wordDocMatrix.rowMatrix);
+ svdMatrix = MatrixUtil.buildSVDMatrix(ifIdfMatrix, svdDimension);
+ this.svdMatrix = svdMatrix;
+ this.wordRDD = RDDUtil.getAllWordsInDoc(docwordRDD);
+ return svdMatrix;
+ }
+
+ /**
+ * Build svd matrix from CSV file.
+ *
+ * @param tfidfCSVfile tf-idf matrix csv file
+ * @param svdDimension: Dimension of matrix after singular value decomposition
+ * @return row matrix
+ */
+ public RowMatrix buildSVDMatrix(String tfidfCSVfile, int svdDimension) {
+ RowMatrix svdMatrix = null;
+ JavaPairRDD<String, Vector> tfidfRDD = MatrixUtil.loadVectorFromCSV(spark, tfidfCSVfile, 2);
+ JavaRDD<Vector> vectorRDD = tfidfRDD.values();
+
+ svdMatrix = MatrixUtil.buildSVDMatrix(vectorRDD, svdDimension);
+ this.svdMatrix = svdMatrix;
+
+ this.wordRDD = tfidfRDD.keys();
+
+ return svdMatrix;
+ }
+
+ /**
+ * Calculate similarity
+ */
+ public void calSimilarity() {
+ CoordinateMatrix simMatrix = SimilarityUtil.calculateSimilarityFromMatrix(svdMatrix);
+ this.simMatrix = simMatrix;
+ }
+
+ /**
+ * Insert linkage triples to elasticsearch
+ *
+ * @param index index name
+ * @param type linkage triple name
+ */
+ public void insertLinkageToES(String index, String type) {
+ List<LinkageTriple> triples = SimilarityUtil.matrixToTriples(wordRDD, simMatrix);
+ try {
+ LinkageTriple.insertTriples(es, triples, index, type);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/utils/SimilarityUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/utils/SimilarityUtil.java b/core/src/main/java/org/apache/sdap/mudrod/utils/SimilarityUtil.java
new file mode 100644
index 0000000..6fdc66d
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/utils/SimilarityUtil.java
@@ -0,0 +1,277 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.utils;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.Optional;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.PairFunction;
+import org.apache.spark.mllib.linalg.Vector;
+import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix;
+import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;
+import org.apache.spark.mllib.linalg.distributed.MatrixEntry;
+import org.apache.spark.mllib.linalg.distributed.RowMatrix;
+import scala.Tuple2;
+
+import java.util.List;
+
+/**
+ * Similarity and distrance calculation utilities
+ */
+public class SimilarityUtil {
+
+ public static final int SIM_COSINE = 3;
+ public static final int SIM_HELLINGER = 2;
+ public static final int SIM_PEARSON = 1;
+ /**
+ * CalSimilarityFromMatrix: Calculate term similarity from matrix.
+ *
+ * @param svdMatrix. Each row is corresponding to a term, and each column is
+ * corresponding to a dimension of feature
+ * @return CoordinateMatrix, each row is corresponding to a term, and each
+ * column is also a term, the cell value is the similarity between the
+ * two terms
+ */
+ public static CoordinateMatrix calculateSimilarityFromMatrix(RowMatrix svdMatrix) {
+ JavaRDD<Vector> vecs = svdMatrix.rows().toJavaRDD();
+ return SimilarityUtil.calculateSimilarityFromVector(vecs);
+ }
+
+ /**
+ * CalSimilarityFromVector:Calculate term similarity from vector.
+ *
+ * @param vecs Each vector is corresponding to a term in the feature space.
+ * @return CoordinateMatrix, each row is corresponding to a term, and each
+ * column is also a term, the cell value is the similarity between the
+ * two terms
+ */
+ public static CoordinateMatrix calculateSimilarityFromVector(JavaRDD<Vector> vecs) {
+ IndexedRowMatrix indexedMatrix = MatrixUtil.buildIndexRowMatrix(vecs);
+ RowMatrix transposeMatrix = MatrixUtil.transposeMatrix(indexedMatrix);
+ return transposeMatrix.columnSimilarities();
+ }
+
+ /**
+ * Calculate term similarity from vector.
+ *
+ * @param importRDD the {@link org.apache.spark.api.java.JavaPairRDD}
+ * data structure containing the vectors.
+ * @param simType the similarity calculation to execute e.g.
+ * <ul>
+ * <li>{@link org.apache.sdap.mudrod.utils.SimilarityUtil#SIM_COSINE} - 3,</li>
+ * <li>{@link org.apache.sdap.mudrod.utils.SimilarityUtil#SIM_HELLINGER} - 2,</li>
+ * <li>{@link org.apache.sdap.mudrod.utils.SimilarityUtil#SIM_PEARSON} - 1</li>
+ * </ul>
+ * @return a new {@link org.apache.spark.api.java.JavaPairRDD}
+ */
+ public static JavaRDD<LinkageTriple> calculateSimilarityFromVector(JavaPairRDD<String, Vector> importRDD, int simType) {
+ JavaRDD<Tuple2<String, Vector>> importRDD1 = importRDD.map(f -> new Tuple2<String, Vector>(f._1, f._2));
+ JavaPairRDD<Tuple2<String, Vector>, Tuple2<String, Vector>> cartesianRDD = importRDD1.cartesian(importRDD1);
+
+ return cartesianRDD.map(new Function<Tuple2<Tuple2<String, Vector>, Tuple2<String, Vector>>, LinkageTriple>() {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public LinkageTriple call(Tuple2<Tuple2<String, Vector>, Tuple2<String, Vector>> arg) {
+ String keyA = arg._1._1;
+ String keyB = arg._2._1;
+
+ if (keyA.equals(keyB)) {
+ return null;
+ }
+
+ Vector vecA = arg._1._2;
+ Vector vecB = arg._2._2;
+ Double weight = 0.0;
+
+ if (simType == SimilarityUtil.SIM_PEARSON) {
+ weight = SimilarityUtil.pearsonDistance(vecA, vecB);
+ } else if (simType == SimilarityUtil.SIM_HELLINGER) {
+ weight = SimilarityUtil.hellingerDistance(vecA, vecB);
+ }
+
+ LinkageTriple triple = new LinkageTriple();
+ triple.keyA = keyA;
+ triple.keyB = keyB;
+ triple.weight = weight;
+ return triple;
+ }
+ }).filter(new Function<LinkageTriple, Boolean>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Boolean call(LinkageTriple arg0) throws Exception {
+ if (arg0 == null) {
+ return false;
+ }
+ return true;
+ }
+ });
+ }
+
+ /**
+ * MatrixtoTriples:Convert term similarity matrix to linkage triple list.
+ *
+ * @param keys each key is a term
+ * @param simMatirx term similarity matrix, in which each row and column is a term and
+ * the cell value is the similarity between the two terms
+ * @return linkage triple list
+ */
+ public static List<LinkageTriple> matrixToTriples(JavaRDD<String> keys, CoordinateMatrix simMatirx) {
+ if (simMatirx.numCols() != keys.count()) {
+ return null;
+ }
+
+ // index words
+ JavaPairRDD<Long, String> keyIdRDD = JavaPairRDD.fromJavaRDD(keys.zipWithIndex().map(new Function<Tuple2<String, Long>, Tuple2<Long, String>>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<Long, String> call(Tuple2<String, Long> docId) {
+ return docId.swap();
+ }
+ }));
+
+ JavaPairRDD<Long, LinkageTriple> entriesRowRDD = simMatirx.entries().toJavaRDD().mapToPair(new PairFunction<MatrixEntry, Long, LinkageTriple>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<Long, LinkageTriple> call(MatrixEntry t) throws Exception {
+ LinkageTriple triple = new LinkageTriple();
+ triple.keyAId = t.i();
+ triple.keyBId = t.j();
+ triple.weight = t.value();
+ return new Tuple2<>(triple.keyAId, triple);
+ }
+ });
+
+ JavaPairRDD<Long, LinkageTriple> entriesColRDD = entriesRowRDD.leftOuterJoin(keyIdRDD).values().mapToPair(new PairFunction<Tuple2<LinkageTriple, Optional<String>>, Long, LinkageTriple>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<Long, LinkageTriple> call(Tuple2<LinkageTriple, Optional<String>> t) throws Exception {
+ LinkageTriple triple = t._1;
+ Optional<String> stra = t._2;
+ if (stra.isPresent()) {
+ triple.keyA = stra.get();
+ }
+ return new Tuple2<>(triple.keyBId, triple);
+ }
+ });
+
+ JavaRDD<LinkageTriple> tripleRDD = entriesColRDD.leftOuterJoin(keyIdRDD).values().map(new Function<Tuple2<LinkageTriple, Optional<String>>, LinkageTriple>() {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public LinkageTriple call(Tuple2<LinkageTriple, Optional<String>> t) throws Exception {
+ LinkageTriple triple = t._1;
+ Optional<String> strb = t._2;
+ if (strb.isPresent()) {
+ triple.keyB = strb.get();
+ }
+ return triple;
+ }
+ });
+ return tripleRDD.collect();
+ }
+
+ /**
+ * Calculate similarity (Hellinger Distance) between vectors
+ *
+ * @param vecA initial vector from which to calculate a similarity
+ * @param vecB second vector involved in similarity calculation
+ * @return similarity between two vectors
+ */
+ public static double hellingerDistance(Vector vecA, Vector vecB) {
+ double[] arrA = vecA.toArray();
+ double[] arrB = vecB.toArray();
+
+ double sim = 0.0;
+
+ int arrsize = arrA.length;
+ for (int i = 0; i < arrsize; i++) {
+ double a = arrA[i];
+ double b = arrB[i];
+ double sqrtDiff = Math.sqrt(a) - Math.sqrt(b);
+ sim += sqrtDiff * sqrtDiff;
+ }
+
+ sim = sim / Math.sqrt(2);
+
+ return sim;
+ }
+
+ /**
+ * Calculate similarity (Pearson Distance) between vectors
+ *
+ * @param vecA initial vector from which to calculate a similarity
+ * @param vecB second vector involved in similarity calculation
+ * @return similarity between two vectors
+ */
+ public static double pearsonDistance(Vector vecA, Vector vecB) {
+ double[] arrA = vecA.toArray();
+ double[] arrB = vecB.toArray();
+
+ int viewA = 0;
+ int viewB = 0;
+ int viewAB = 0;
+
+ int arrsize = arrA.length;
+ for (int i = 0; i < arrsize; i++) {
+ if (arrA[i] > 0) {
+ viewA++;
+ }
+
+ if (arrB[i] > 0) {
+ viewB++;
+ }
+
+ if (arrB[i] > 0 && arrA[i] > 0) {
+ viewAB++;
+ }
+ }
+ return viewAB / (Math.sqrt(viewA) * Math.sqrt(viewB));
+ }
+
+ /**
+ * calculate similarity between vectors
+ *
+ * @param vecA initial vector from which to calculate a similarity
+ * @param vecB second vector involved in similarity calculation
+ * @return similarity between two vectors
+ */
+ public static double cosineDistance(Vector vecA, Vector vecB) {
+ return 1;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/utils/package-info.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/utils/package-info.java b/core/src/main/java/org/apache/sdap/mudrod/utils/package-info.java
new file mode 100644
index 0000000..1adb0b9
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/utils/package-info.java
@@ -0,0 +1,18 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * This package includes utilities classes for calculating similarity and
+ * parsing HTTP request
+ */
+package org.apache.sdap.mudrod.utils;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/package-info.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/package-info.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/package-info.java
new file mode 100644
index 0000000..9c87a7d
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/package-info.java
@@ -0,0 +1,18 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * This package includes web log pre-processing, processing, and data structure
+ * classes.
+ */
+package org.apache.sdap.mudrod.weblog;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/KGreedyPartitionSolver.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/KGreedyPartitionSolver.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/KGreedyPartitionSolver.java
new file mode 100644
index 0000000..8f4e263
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/KGreedyPartitionSolver.java
@@ -0,0 +1,142 @@
+package org.apache.sdap.mudrod.weblog.partition;
+
+import java.util.*;
+
+public class KGreedyPartitionSolver implements ThePartitionProblemSolver {
+
+ public boolean bsorted = false;
+
+ public KGreedyPartitionSolver() {
+ // default constructor
+ }
+
+ public KGreedyPartitionSolver(boolean bsorted) {
+ this.bsorted = true;
+ }
+
+ @Override
+ public Map<String, Integer> solve(Map<String, Double> labelNums, int k) {
+ List<Double> lista = null;
+ List<String> months = null;
+
+ if (!this.bsorted) {
+ LinkedHashMap sortedMap = this.sortMapByValue(labelNums);
+ lista = new ArrayList(sortedMap.values());
+ months = new ArrayList(sortedMap.keySet());
+ } else {
+ lista = new ArrayList(labelNums.values());
+ months = new ArrayList(labelNums.keySet());
+ }
+
+ List<List<Double>> parts = new ArrayList<>();
+ List<List<String>> splitMonths = new ArrayList<>();
+
+ for (int i = 0; i < k; i++) {
+ List<Double> part = new ArrayList();
+ parts.add(part);
+
+ List<String> monthList = new ArrayList();
+ splitMonths.add(monthList);
+ }
+
+ int j = 0;
+ for (Double lista1 : lista) {
+
+ Double minimalSum = -1.0;
+ int position = 0;
+ for (int i = 0; i < parts.size(); i++) {
+ List<Double> part = parts.get(i);
+ if (minimalSum == -1) {
+ minimalSum = Suma(part);
+ position = i;
+ } else if (Suma(part) < minimalSum) {
+ minimalSum = Suma(part);
+ position = i;
+ }
+ }
+
+ List<Double> part = parts.get(position);
+ part.add(lista1);
+ parts.set(position, part);
+
+ List<String> month = splitMonths.get(position);
+ month.add(months.get(j));
+ splitMonths.set(position, month);
+ j++;
+ }
+
+ /* for(int i=0; i<splitMonths.size(); i++){
+ System.out.println("group:" + i);
+ printStrList(splitMonths.get(i));
+ }
+
+ for(int i=0; i<parts.size(); i++){
+ print(parts.get(i));
+ }*/
+
+ Map<String, Integer> LabelGroups = new HashMap<String, Integer>();
+ for (int i = 0; i < splitMonths.size(); i++) {
+ List<String> list = splitMonths.get(i);
+ for (int m = 0; m < list.size(); m++) {
+ LabelGroups.put(list.get(m), i);
+ }
+ }
+
+ return LabelGroups;
+ }
+
+ public LinkedHashMap<String, Double> sortMapByValue(Map passedMap) {
+ List mapKeys = new ArrayList(passedMap.keySet());
+ List mapValues = new ArrayList(passedMap.values());
+ Collections.sort(mapValues, Collections.reverseOrder());
+ Collections.sort(mapKeys, Collections.reverseOrder());
+
+ LinkedHashMap sortedMap = new LinkedHashMap();
+
+ Iterator valueIt = mapValues.iterator();
+ while (valueIt.hasNext()) {
+ Object val = valueIt.next();
+ Iterator keyIt = mapKeys.iterator();
+
+ while (keyIt.hasNext()) {
+ Object key = keyIt.next();
+ String comp1 = passedMap.get(key).toString();
+ String comp2 = val.toString();
+
+ if (comp1.equals(comp2)) {
+ passedMap.remove(key);
+ mapKeys.remove(key);
+ sortedMap.put((String) key, (Double) val);
+ break;
+ }
+
+ }
+
+ }
+ return sortedMap;
+ }
+
+ private Double Suma(List<Double> part) {
+ Double ret = 0.0;
+ for (int i = 0; i < part.size(); i++) {
+ ret += part.get(i);
+ }
+ return ret;
+ }
+
+ private void print(List<Double> list) {
+ /*for (int i = 0; i < list.size(); i++) {
+ System.out.print(list.get(i)+",");
+ }*/
+ System.out.print("sum is:" + Suma(list));
+ System.out.println();
+ }
+
+ private void printStrList(List<String> list) {
+ for (int i = 0; i < list.size(); i++) {
+ System.out.print(list.get(i) + ",");
+ }
+ System.out.println();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/ThePartitionProblemSolver.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/ThePartitionProblemSolver.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/ThePartitionProblemSolver.java
new file mode 100644
index 0000000..507140f
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/ThePartitionProblemSolver.java
@@ -0,0 +1,8 @@
+package org.apache.sdap.mudrod.weblog.partition;
+
+import java.util.Map;
+
+public interface ThePartitionProblemSolver {
+
+ public Map<String, Integer> solve(Map<String, Double> labelNums, int k);
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/logPartitioner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/logPartitioner.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/logPartitioner.java
new file mode 100644
index 0000000..7ff2181
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/partition/logPartitioner.java
@@ -0,0 +1,33 @@
+package org.apache.sdap.mudrod.weblog.partition;
+
+import org.apache.spark.Partitioner;
+
+import java.util.Map;
+
+public class logPartitioner extends Partitioner {
+
+ int num;
+ Map<String, Integer> UserGroups;
+
+ public logPartitioner(int num) {
+ this.num = num;
+ }
+
+ public logPartitioner(Map<String, Integer> UserGroups, int num) {
+ this.UserGroups = UserGroups;
+ this.num = num;
+ }
+
+ @Override
+ public int getPartition(Object arg0) {
+ // TODO Auto-generated method stub
+ String user = (String) arg0;
+ return UserGroups.get(user);
+ }
+
+ @Override
+ public int numPartitions() {
+ // TODO Auto-generated method stub
+ return num;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ClickStreamGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ClickStreamGenerator.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ClickStreamGenerator.java
new file mode 100644
index 0000000..e678854
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ClickStreamGenerator.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.sdap.mudrod.discoveryengine.DiscoveryStepAbstract;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.utils.LabeledRowMatrix;
+import org.apache.sdap.mudrod.utils.MatrixUtil;
+import org.apache.sdap.mudrod.weblog.structure.ClickStream;
+import org.apache.sdap.mudrod.weblog.structure.SessionExtractor;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Supports ability to extract click stream data based on session processing results
+ */
+public class ClickStreamGenerator extends DiscoveryStepAbstract {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(ClickStreamGenerator.class);
+
+ public ClickStreamGenerator(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ }
+
+ @Override
+ public Object execute() {
+ LOG.info("Starting ClickStreamGenerator...");
+ startTime = System.currentTimeMillis();
+
+ String clickstremMatrixFile = props.getProperty("clickstreamMatrix");
+ try {
+ SessionExtractor extractor = new SessionExtractor();
+ JavaRDD<ClickStream> clickstreamRDD = extractor.extractClickStreamFromES(this.props, this.es, this.spark);
+ int weight = Integer.parseInt(props.getProperty("downloadWeight"));
+ JavaPairRDD<String, List<String>> metaddataQueryRDD = extractor.bulidDataQueryRDD(clickstreamRDD, weight);
+ LabeledRowMatrix wordDocMatrix = MatrixUtil.createWordDocMatrix(metaddataQueryRDD);
+
+ MatrixUtil.exportToCSV(wordDocMatrix.rowMatrix, wordDocMatrix.rowkeys, wordDocMatrix.colkeys, clickstremMatrixFile);
+ } catch (Exception e) {
+ LOG.error("Encountered error within ClickStreamGenerator: {}", e);
+ }
+
+ endTime = System.currentTimeMillis();
+ LOG.info("ClickStreamGenerator complete. Time elapsed {} seconds.", (endTime - startTime) / 1000);
+ return null;
+ }
+
+ @Override
+ public Object execute(Object o) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/CrawlerDetection.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/CrawlerDetection.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/CrawlerDetection.java
new file mode 100644
index 0000000..79a014e
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/CrawlerDetection.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.sdap.mudrod.discoveryengine.DiscoveryStepAbstract;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.main.MudrodConstants;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.Function2;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Order;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.joda.time.DateTime;
+import org.joda.time.Seconds;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.ISODateTimeFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * An {@link DiscoveryStepAbstract}
+ * implementation which detects a known list of Web crawlers which may may be
+ * present within, and pollute various logs acting as input to Mudrod.
+ */
+public class CrawlerDetection extends LogAbstract {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(CrawlerDetection.class);
+
+ public static final String CRAWLER = "crawler";
+ public static final String GOOGLE_BOT = "googlebot";
+ public static final String BING_BOT = "bingbot";
+ public static final String YAHOO_BOT = "slurp";
+ public static final String YACY_BOT = "yacybot";
+ public static final String ROGER_BOT = "rogerbot";
+ public static final String YANDEX_BOT = "yandexbot";
+
+ public static final String NO_AGENT_BOT = "-";
+ public static final String PERL_BOT = "libwww-perl/";
+ public static final String APACHE_HHTP = "apache-httpclient/";
+ public static final String JAVA_CLIENT = "java/";
+ public static final String CURL = "curl/";
+
+ /**
+ * Paramterized constructor to instantiate a configured instance of
+ * {@link CrawlerDetection}
+ *
+ * @param props populated {@link java.util.Properties} object
+ * @param es {@link ESDriver} object to use in
+ * crawler detection preprocessing.
+ * @param spark {@link SparkDriver} object to use in
+ * crawler detection preprocessing.
+ */
+ public CrawlerDetection(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ }
+
+ public CrawlerDetection() {
+ super(null, null, null);
+ }
+
+ @Override
+ public Object execute() {
+ LOG.info("Starting Crawler detection {}.", httpType);
+ startTime = System.currentTimeMillis();
+ try {
+ checkByRate();
+ } catch (InterruptedException | IOException e) {
+ LOG.error("Encountered an error whilst detecting Web crawlers.", e);
+ }
+ endTime = System.currentTimeMillis();
+ es.refreshIndex();
+ LOG.info("Crawler detection complete. Time elapsed {} seconds", (endTime - startTime) / 1000);
+ return null;
+ }
+
+ /**
+ * Check known crawler through crawler agent name list
+ *
+ * @param agent name of a log line
+ * @return 1 if the log is initiated by crawler, 0 otherwise
+ */
+ public boolean checkKnownCrawler(String agent) {
+ agent = agent.toLowerCase();
+ if (agent.contains(CRAWLER) || agent.contains(GOOGLE_BOT) || agent.contains(BING_BOT) || agent.contains(APACHE_HHTP) || agent.contains(PERL_BOT) || agent.contains(YAHOO_BOT) || agent
+ .contains(YANDEX_BOT) || agent.contains(NO_AGENT_BOT) || agent.contains(PERL_BOT) || agent.contains(APACHE_HHTP) || agent.contains(JAVA_CLIENT) || agent.contains(CURL)) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public void checkByRate() throws InterruptedException, IOException {
+ String processingType = props.getProperty(MudrodConstants.PROCESS_TYPE);
+ if (processingType.equals("sequential")) {
+ checkByRateInSequential();
+ } else if (processingType.equals("parallel")) {
+ checkByRateInParallel();
+ }
+ }
+
+ /**
+ * Check crawler by request sending rate, which is read from configruation
+ * file
+ *
+ * @throws InterruptedException InterruptedException
+ * @throws IOException IOException
+ */
+ public void checkByRateInSequential() throws InterruptedException, IOException {
+ es.createBulkProcessor();
+
+ int rate = Integer.parseInt(props.getProperty("sendingrate"));
+
+ Terms users = this.getUserTerms(this.httpType);
+ LOG.info("Original User count: {}", Integer.toString(users.getBuckets().size()));
+
+ int userCount = 0;
+ for (Terms.Bucket entry : users.getBuckets()) {
+ String user = entry.getKey().toString();
+ int count = checkByRate(es, user);
+ userCount += count;
+ }
+ es.destroyBulkProcessor();
+ LOG.info("User count: {}", Integer.toString(userCount));
+ }
+
+ void checkByRateInParallel() throws InterruptedException, IOException {
+
+ JavaRDD<String> userRDD = getUserRDD(this.httpType);
+ LOG.info("Original User count: {}", userRDD.count());
+
+ int userCount = 0;
+ userCount = userRDD.mapPartitions((FlatMapFunction<Iterator<String>, Integer>) iterator -> {
+ ESDriver tmpES = new ESDriver(props);
+ tmpES.createBulkProcessor();
+ List<Integer> realUserNums = new ArrayList<>();
+ while (iterator.hasNext()) {
+ String s = iterator.next();
+ Integer realUser = checkByRate(tmpES, s);
+ realUserNums.add(realUser);
+ }
+ tmpES.destroyBulkProcessor();
+ tmpES.close();
+ return realUserNums.iterator();
+ }).reduce((Function2<Integer, Integer, Integer>) (a, b) -> a + b);
+
+ LOG.info("User count: {}", Integer.toString(userCount));
+ }
+
+ private int checkByRate(ESDriver es, String user) {
+
+ int rate = Integer.parseInt(props.getProperty("sendingrate"));
+ Pattern pattern = Pattern.compile("get (.*?) http/*");
+ Matcher matcher;
+
+ BoolQueryBuilder filterSearch = new BoolQueryBuilder();
+ filterSearch.must(QueryBuilders.termQuery("IP", user));
+
+ AggregationBuilder aggregation = AggregationBuilders.dateHistogram("by_minute").field("Time").dateHistogramInterval(DateHistogramInterval.MINUTE).order(Order.COUNT_DESC);
+ SearchResponse checkRobot = es.getClient().prepareSearch(logIndex).setTypes(httpType, ftpType).setQuery(filterSearch).setSize(0).addAggregation(aggregation).execute().actionGet();
+
+ Histogram agg = checkRobot.getAggregations().get("by_minute");
+
+ List<? extends Histogram.Bucket> botList = agg.getBuckets();
+ long maxCount = botList.get(0).getDocCount();
+ if (maxCount >= rate) {
+ return 0;
+ } else {
+ DateTime dt1 = null;
+ int toLast = 0;
+ SearchResponse scrollResp = es.getClient().prepareSearch(logIndex).setTypes(httpType, ftpType).setScroll(new TimeValue(60000)).setQuery(filterSearch).setSize(100).execute().actionGet();
+ while (true) {
+ for (SearchHit hit : scrollResp.getHits().getHits()) {
+ Map<String, Object> result = hit.getSource();
+ String logtype = (String) result.get("LogType");
+ if (logtype.equals("PO.DAAC")) {
+ String request = (String) result.get("Request");
+ matcher = pattern.matcher(request.trim().toLowerCase());
+ boolean find = false;
+ while (matcher.find()) {
+ request = matcher.group(1);
+ result.put("RequestUrl", "http://podaac.jpl.nasa.gov" + request);
+ find = true;
+ }
+ if (!find) {
+ result.put("RequestUrl", request);
+ }
+ } else {
+ result.put("RequestUrl", result.get("Request"));
+ }
+
+ DateTimeFormatter fmt = ISODateTimeFormat.dateTime();
+ DateTime dt2 = fmt.parseDateTime((String) result.get("Time"));
+
+ if (dt1 == null) {
+ toLast = 0;
+ } else {
+ toLast = Math.abs(Seconds.secondsBetween(dt1, dt2).getSeconds());
+ }
+ result.put("ToLast", toLast);
+ IndexRequest ir = new IndexRequest(logIndex, cleanupType).source(result);
+
+ es.getBulkProcessor().add(ir);
+ dt1 = dt2;
+ }
+
+ scrollResp = es.getClient().prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(600000)).execute().actionGet();
+ if (scrollResp.getHits().getHits().length == 0) {
+ break;
+ }
+ }
+
+ }
+
+ return 1;
+ }
+
+ @Override
+ public Object execute(Object o) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/HistoryGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/HistoryGenerator.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/HistoryGenerator.java
new file mode 100644
index 0000000..f92d79c
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/HistoryGenerator.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.main.MudrodConstants;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Supports ability to generate search history (queries) for each individual
+ * user (IP)
+ */
+public class HistoryGenerator extends LogAbstract {
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(HistoryGenerator.class);
+
+ public HistoryGenerator(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ }
+
+ @Override
+ public Object execute() {
+ LOG.info("Starting HistoryGenerator...");
+ startTime = System.currentTimeMillis();
+
+ generateBinaryMatrix();
+
+ endTime = System.currentTimeMillis();
+ LOG.info("HistoryGenerator complete. Time elapsed {} seconds", (endTime - startTime) / 1000);
+ return null;
+ }
+
+ /**
+ * Method to generate a binary user*query matrix (stored in temporary .csv
+ * file)
+ */
+ public void generateBinaryMatrix() {
+ try {
+
+ System.out.println(props.getProperty("userHistoryMatrix"));
+ File file = new File(props.getProperty("userHistoryMatrix"));
+ if (file.exists()) {
+ file.delete();
+ }
+
+ file.createNewFile();
+
+ FileWriter fw = new FileWriter(file.getAbsoluteFile());
+ BufferedWriter bw = new BufferedWriter(fw);
+ bw.write("Num" + ",");
+
+ // step 1: write first row of csv
+ List<String> logIndexList = es.getIndexListWithPrefix(props.getProperty(MudrodConstants.LOG_INDEX));
+
+ String[] logIndices = logIndexList.toArray(new String[0]);
+ String[] statictypeArray = new String[] { this.sessionStats };
+ int docCount = es.getDocCount(logIndices, statictypeArray);
+
+ SearchResponse sr = es.getClient().prepareSearch(logIndices).setTypes(statictypeArray).setQuery(QueryBuilders.matchAllQuery()).setSize(0)
+ .addAggregation(AggregationBuilders.terms("IPs").field("IP").size(docCount)).execute().actionGet();
+ Terms ips = sr.getAggregations().get("IPs");
+ List<String> ipList = new ArrayList<>();
+ for (Terms.Bucket entry : ips.getBuckets()) {
+ if (entry.getDocCount() > Integer.parseInt(props.getProperty(MudrodConstants.MINI_USER_HISTORY))) { // filter
+ // out
+ // less
+ // active users/ips
+ ipList.add(entry.getKey().toString());
+ }
+ }
+ bw.write(String.join(",", ipList) + "\n");
+
+ // step 2: step the rest rows of csv
+ SearchRequestBuilder sr2Builder = es.getClient().prepareSearch(logIndices).setTypes(statictypeArray).setQuery(QueryBuilders.matchAllQuery()).setSize(0)
+ .addAggregation(AggregationBuilders.terms("KeywordAgg").field("keywords").size(docCount).subAggregation(AggregationBuilders.terms("IPAgg").field("IP").size(docCount)));
+
+ SearchResponse sr2 = sr2Builder.execute().actionGet();
+ Terms keywords = sr2.getAggregations().get("KeywordAgg");
+
+ for (Terms.Bucket keyword : keywords.getBuckets()) {
+
+ Map<String, Integer> ipMap = new HashMap<>();
+ Terms ipAgg = keyword.getAggregations().get("IPAgg");
+
+ int distinctUser = ipAgg.getBuckets().size();
+ if (distinctUser > Integer.parseInt(props.getProperty(MudrodConstants.MINI_USER_HISTORY))) {
+ bw.write(keyword.getKey() + ",");
+ for (Terms.Bucket IP : ipAgg.getBuckets()) {
+
+ ipMap.put(IP.getKey().toString(), 1);
+ }
+ for (int i = 0; i < ipList.size(); i++) {
+ if (ipMap.containsKey(ipList.get(i))) {
+ bw.write(ipMap.get(ipList.get(i)) + ",");
+ } else {
+ bw.write("0,");
+ }
+ }
+ bw.write("\n");
+ }
+ }
+
+ bw.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+
+ }
+
+ @Override
+ public Object execute(Object o) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ImportLogFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ImportLogFile.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ImportLogFile.java
new file mode 100644
index 0000000..ca47f01
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ImportLogFile.java
@@ -0,0 +1,343 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.main.MudrodConstants;
+import org.apache.sdap.mudrod.weblog.structure.ApacheAccessLog;
+import org.apache.sdap.mudrod.weblog.structure.FtpLog;
+import org.apache.spark.api.java.JavaRDD;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+
+/**
+ * Supports ability to parse and process FTP and HTTP log files
+ */
+public class ImportLogFile extends LogAbstract {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ImportLogFile.class);
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ String logEntryPattern = "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] " + "\"(.+?)\" (\\d{3}) (\\d+|-) \"((?:[^\"]|\")+)\" \"([^\"]+)\"";
+
+ public static final int NUM_FIELDS = 9;
+ Pattern p = Pattern.compile(logEntryPattern);
+ transient Matcher matcher;
+
+ /**
+ * Constructor supporting a number of parameters documented below.
+ *
+ * @param props a {@link java.util.Map} containing K,V of type String, String
+ * respectively.
+ * @param es the {@link ESDriver} used to persist log
+ * files.
+ * @param spark the {@link SparkDriver} used to process
+ * input log files.
+ */
+ public ImportLogFile(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ }
+
+ @Override
+ public Object execute() {
+ LOG.info("Starting Log Import {}", props.getProperty(MudrodConstants.TIME_SUFFIX));
+ startTime = System.currentTimeMillis();
+ readFile();
+ endTime = System.currentTimeMillis();
+ LOG.info("Log Import complete. Time elapsed {} seconds", (endTime - startTime) / 1000);
+ es.refreshIndex();
+ return null;
+ }
+
+ /**
+ * Utility function to aid String to Number formatting such that three letter
+ * months such as 'Jan' are converted to the Gregorian integer equivalent.
+ *
+ * @param time the input {@link java.lang.String} to convert to int.
+ * @return the converted Month as an int.
+ */
+ public String switchtoNum(String time) {
+ String newTime = time;
+ if (newTime.contains("Jan")) {
+ newTime = newTime.replace("Jan", "1");
+ } else if (newTime.contains("Feb")) {
+ newTime = newTime.replace("Feb", "2");
+ } else if (newTime.contains("Mar")) {
+ newTime = newTime.replace("Mar", "3");
+ } else if (newTime.contains("Apr")) {
+ newTime = newTime.replace("Apr", "4");
+ } else if (newTime.contains("May")) {
+ newTime = newTime.replace("May", "5");
+ } else if (newTime.contains("Jun")) {
+ newTime = newTime.replace("Jun", "6");
+ } else if (newTime.contains("Jul")) {
+ newTime = newTime.replace("Jul", "7");
+ } else if (newTime.contains("Aug")) {
+ newTime = newTime.replace("Aug", "8");
+ } else if (newTime.contains("Sep")) {
+ newTime = newTime.replace("Sep", "9");
+ } else if (newTime.contains("Oct")) {
+ newTime = newTime.replace("Oct", "10");
+ } else if (newTime.contains("Nov")) {
+ newTime = newTime.replace("Nov", "11");
+ } else if (newTime.contains("Dec")) {
+ newTime = newTime.replace("Dec", "12");
+ }
+ return newTime;
+ }
+
+ public void readFile() {
+
+ String httplogpath = null;
+ String ftplogpath = null;
+
+ File directory = new File(props.getProperty(MudrodConstants.DATA_DIR));
+ File[] fList = directory.listFiles();
+ for (File file : fList) {
+ if (file.isFile() && file.getName().contains(props.getProperty(MudrodConstants.TIME_SUFFIX)))
+ {
+ if (file.getName().contains(props.getProperty(MudrodConstants.HTTP_PREFIX)))
+ {
+ httplogpath = file.getAbsolutePath();
+ }
+
+ if (file.getName().contains(props.getProperty(MudrodConstants.FTP_PREFIX)))
+ {
+ ftplogpath = file.getAbsolutePath();
+ }
+ }
+ }
+
+ if(httplogpath == null || ftplogpath == null)
+ {
+ LOG.error("WWW file or FTP logs cannot be found, please check your data directory.");
+ return;
+ }
+
+ String processingType = props.getProperty(MudrodConstants.PROCESS_TYPE, "parallel");
+ if (processingType.equals("sequential")) {
+ readFileInSequential(httplogpath, ftplogpath);
+ } else if (processingType.equals("parallel")) {
+ readFileInParallel(httplogpath, ftplogpath);
+ }
+ }
+
+ /**
+ * Read the FTP or HTTP log path with the intention of processing lines from
+ * log files.
+ *
+ * @param httplogpath path to the parent directory containing http logs
+ * @param ftplogpath path to the parent directory containing ftp logs
+ */
+ public void readFileInSequential(String httplogpath, String ftplogpath) {
+ es.createBulkProcessor();
+ try {
+ readLogFile(httplogpath, "http", logIndex, httpType);
+ readLogFile(ftplogpath, "FTP", logIndex, ftpType);
+
+ } catch (IOException e) {
+ LOG.error("Error whilst reading log file.", e);
+ }
+ es.destroyBulkProcessor();
+ }
+
+ /**
+ * Read the FTP or HTTP log path with the intention of processing lines from
+ * log files.
+ *
+ * @param httplogpath path to the parent directory containing http logs
+ * @param ftplogpath path to the parent directory containing ftp logs
+ */
+ public void readFileInParallel(String httplogpath, String ftplogpath) {
+
+ importHttpfile(httplogpath);
+ importFtpfile(ftplogpath);
+ }
+
+ public void importHttpfile(String httplogpath) {
+ // import http logs
+ JavaRDD<String> accessLogs = spark.sc.textFile(httplogpath, this.partition).map(s -> ApacheAccessLog.parseFromLogLine(s)).filter(ApacheAccessLog::checknull);
+
+ JavaEsSpark.saveJsonToEs(accessLogs, logIndex + "/" + this.httpType);
+ }
+
+ public void importFtpfile(String ftplogpath) {
+ // import ftp logs
+ JavaRDD<String> ftpLogs = spark.sc.textFile(ftplogpath, this.partition).map(s -> FtpLog.parseFromLogLine(s)).filter(FtpLog::checknull);
+
+ JavaEsSpark.saveJsonToEs(ftpLogs, logIndex + "/" + this.ftpType);
+ }
+
+ /**
+ * Process a log path on local file system which contains the relevant
+ * parameters as below.
+ *
+ * @param fileName the {@link java.lang.String} path to the log directory on file
+ * system
+ * @param protocol whether to process 'http' or 'FTP'
+ * @param index the index name to write logs to
+ * @param type one of the available protocols from which Mudrod logs are obtained.
+ * @throws IOException if there is an error reading anything from the fileName provided.
+ */
+ public void readLogFile(String fileName, String protocol, String index, String type) throws IOException {
+ BufferedReader br = new BufferedReader(new FileReader(fileName));
+ int count = 0;
+ try {
+ String line = br.readLine();
+ while (line != null) {
+ if ("FTP".equals(protocol)) {
+ parseSingleLineFTP(line, index, type);
+ } else {
+ parseSingleLineHTTP(line, index, type);
+ }
+ line = br.readLine();
+ count++;
+ }
+ } catch (FileNotFoundException e) {
+ LOG.error("File not found.", e);
+ } catch (IOException e) {
+ LOG.error("Error reading input directory.", e);
+ } finally {
+ br.close();
+ LOG.info("Num of {} entries:\t{}", protocol, count);
+ }
+ }
+
+ /**
+ * Parse a single FTP log entry
+ *
+ * @param log a single log line
+ * @param index the index name we wish to persist the log line to
+ * @param type one of the available protocols from which Mudrod logs are obtained.
+ */
+ public void parseSingleLineFTP(String log, String index, String type) {
+ String ip = log.split(" +")[6];
+
+ String time = log.split(" +")[1] + ":" + log.split(" +")[2] + ":" + log.split(" +")[3] + ":" + log.split(" +")[4];
+
+ time = switchtoNum(time);
+ SimpleDateFormat formatter = new SimpleDateFormat("MM:dd:HH:mm:ss:yyyy");
+ Date date = null;
+ try {
+ date = formatter.parse(time);
+ } catch (ParseException e) {
+ LOG.error("Error whilst parsing the date.", e);
+ }
+ String bytes = log.split(" +")[7];
+
+ String request = log.split(" +")[8].toLowerCase();
+
+ if (!request.contains("/misc/") && !request.contains("readme")) {
+ IndexRequest ir;
+ try {
+ ir = new IndexRequest(index, type)
+ .source(jsonBuilder().startObject().field("LogType", "ftp").field("IP", ip).field("Time", date).field("Request", request).field("Bytes", Long.parseLong(bytes)).endObject());
+ es.getBulkProcessor().add(ir);
+ } catch (NumberFormatException e) {
+ LOG.error("Error whilst processing numbers", e);
+ } catch (IOException e) {
+ LOG.error("IOError whilst adding to the bulk processor.", e);
+ }
+ }
+
+ }
+
+ /**
+ * Parse a single HTTP log entry
+ *
+ * @param log a single log line
+ * @param index the index name we wish to persist the log line to
+ * @param type one of the available protocols from which Mudrod logs are obtained.
+ */
+ public void parseSingleLineHTTP(String log, String index, String type) {
+ matcher = p.matcher(log);
+ if (!matcher.matches() || NUM_FIELDS != matcher.groupCount()) {
+ return;
+ }
+ String time = matcher.group(4);
+ time = switchtoNum(time);
+ SimpleDateFormat formatter = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
+ Date date = null;
+ try {
+ date = formatter.parse(time);
+ } catch (ParseException e) {
+ LOG.error("Error whilst attempting to parse date.", e);
+ }
+
+ String bytes = matcher.group(7);
+ if ("-".equals(bytes)) {
+ bytes = "0";
+ }
+
+ String request = matcher.group(5).toLowerCase();
+ String agent = matcher.group(9);
+ CrawlerDetection crawlerDe = new CrawlerDetection(this.props, this.es, this.spark);
+ if (!crawlerDe.checkKnownCrawler(agent)) {
+ boolean tag = false;
+ String[] mimeTypes = { ".js", ".css", ".jpg", ".png", ".ico", "image_captcha", "autocomplete", ".gif", "/alldata/", "/api/", "get / http/1.1", ".jpeg", "/ws/" };
+ for (int i = 0; i < mimeTypes.length; i++) {
+ if (request.contains(mimeTypes[i])) {
+ tag = true;
+ break;
+ }
+ }
+
+ if (!tag) {
+ IndexRequest ir = null;
+ executeBulkRequest(ir, index, type, matcher, date, bytes);
+ }
+ }
+ }
+
+ private void executeBulkRequest(IndexRequest ir, String index, String type, Matcher matcher, Date date, String bytes) {
+ IndexRequest newIr = ir;
+ try {
+ newIr = new IndexRequest(index, type).source(
+ jsonBuilder().startObject().field("LogType", "PO.DAAC").field("IP", matcher.group(1)).field("Time", date).field("Request", matcher.group(5)).field("Response", matcher.group(6))
+ .field("Bytes", Integer.parseInt(bytes)).field("Referer", matcher.group(8)).field("Browser", matcher.group(9)).endObject());
+
+ es.getBulkProcessor().add(newIr);
+ } catch (NumberFormatException e) {
+ LOG.error("Error whilst processing numbers", e);
+ } catch (IOException e) {
+ LOG.error("IOError whilst adding to the bulk processor.", e);
+ }
+ }
+
+ @Override
+ public Object execute(Object o) {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/LogAbstract.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/LogAbstract.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/LogAbstract.java
new file mode 100644
index 0000000..23ddbee
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/LogAbstract.java
@@ -0,0 +1,228 @@
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.sdap.mudrod.discoveryengine.DiscoveryStepAbstract;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.main.MudrodConstants;
+import org.apache.sdap.mudrod.weblog.partition.KGreedyPartitionSolver;
+import org.apache.sdap.mudrod.weblog.partition.ThePartitionProblemSolver;
+import org.apache.sdap.mudrod.weblog.partition.logPartitioner;
+import org.apache.spark.Partition;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Order;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Tuple2;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.*;
+
+public class LogAbstract extends DiscoveryStepAbstract {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(LogAbstract.class);
+
+ public String logIndex = null;
+ public String httpType = null;
+ public String ftpType = null;
+ public String cleanupType = null;
+ public String sessionStats = null;
+ public int partition = 96;
+
+ public LogAbstract(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ if (props != null) {
+ initLogIndex();
+ }
+ }
+
+ protected void initLogIndex() {
+ logIndex = props.getProperty(MudrodConstants.LOG_INDEX) + props.getProperty(MudrodConstants.TIME_SUFFIX);
+ httpType = props.getProperty(MudrodConstants.HTTP_TYPE_PREFIX);
+ ftpType = props.getProperty(MudrodConstants.FTP_TYPE_PREFIX);
+ cleanupType = props.getProperty(MudrodConstants.CLEANUP_TYPE_PREFIX);
+ sessionStats = props.getProperty(MudrodConstants.SESSION_STATS_PREFIX);
+
+ InputStream settingsStream = getClass().getClassLoader().getResourceAsStream(ES_SETTINGS);
+ InputStream mappingsStream = getClass().getClassLoader().getResourceAsStream(ES_MAPPINGS);
+ JSONObject settingsJSON = null;
+ JSONObject mappingJSON = null;
+
+ try {
+ settingsJSON = new JSONObject(IOUtils.toString(settingsStream));
+ } catch (JSONException | IOException e1) {
+ LOG.error("Error reading Elasticsearch settings!", e1);
+ }
+
+ try {
+ mappingJSON = new JSONObject(IOUtils.toString(mappingsStream));
+ } catch (JSONException | IOException e1) {
+ LOG.error("Error reading Elasticsearch mappings!", e1);
+ }
+
+ try {
+ if (settingsJSON != null && mappingJSON != null) {
+ this.es.putMapping(logIndex, settingsJSON.toString(), mappingJSON.toString());
+ }
+ } catch (IOException e) {
+ LOG.error("Error entering Elasticsearch Mappings!", e);
+ }
+ }
+
+ @Override
+ public Object execute() {
+ return null;
+ }
+
+ @Override
+ public Object execute(Object o) {
+ return null;
+ }
+
+ public JavaRDD<String> getUserRDD(String... type) {
+ Map<String, Double> userDocs = getUserDocs(type);
+ return parallizeUsers(userDocs);
+ }
+
+ public List<String> getUsers(String type) {
+
+ Terms users = this.getUserTerms(type);
+ List<String> userList = new ArrayList<>();
+ for (Terms.Bucket entry : users.getBuckets()) {
+ String ip = (String) entry.getKey();
+ userList.add(ip);
+ }
+
+ return userList;
+ }
+
+ public Terms getUserTerms(String... type) {
+
+ int docCount = es.getDocCount(logIndex, type);
+
+ SearchResponse sr = es.getClient().prepareSearch(logIndex).setTypes(type).setQuery(QueryBuilders.matchAllQuery()).setSize(0)
+ .addAggregation(AggregationBuilders.terms("Users").field("IP").size(docCount)).execute().actionGet();
+ return sr.getAggregations().get("Users");
+ }
+
+ public Map<String, Double> getUserDocs(String... type) {
+
+ Terms users = this.getUserTerms(type);
+ Map<String, Double> userList = new HashMap<>();
+ for (Terms.Bucket entry : users.getBuckets()) {
+ String ip = (String) entry.getKey();
+ Long count = entry.getDocCount();
+ userList.put(ip, Double.valueOf(count));
+ }
+
+ return userList;
+ }
+
+ public Map<String, Long> getUserDailyDocs() {
+
+ int docCount = es.getDocCount(logIndex, httpType);
+
+ AggregationBuilder dailyAgg = AggregationBuilders.dateHistogram("by_day").field("Time").dateHistogramInterval(DateHistogramInterval.DAY).order(Order.COUNT_DESC);
+
+ SearchResponse sr = es.getClient().prepareSearch(logIndex).setTypes(httpType).setQuery(QueryBuilders.matchAllQuery()).setSize(0)
+ .addAggregation(AggregationBuilders.terms("Users").field("IP").size(docCount).subAggregation(dailyAgg)).execute().actionGet();
+ Terms users = sr.getAggregations().get("Users");
+ Map<String, Long> userList = new HashMap<>();
+ for (Terms.Bucket user : users.getBuckets()) {
+ String ip = (String) user.getKey();
+
+ System.out.println(ip);
+
+ Histogram agg = user.getAggregations().get("by_day");
+ List<? extends Histogram.Bucket> dateList = agg.getBuckets();
+ int size = dateList.size();
+ for (int i = 0; i < size; i++) {
+ Long count = dateList.get(i).getDocCount();
+ String date = dateList.get(i).getKey().toString();
+
+ System.out.println(date);
+ System.out.println(count);
+ }
+ }
+
+ return userList;
+ }
+
+ protected void checkUserPartition(JavaRDD<String> userRDD) {
+ System.out.println("hhhhh");
+ List<Partition> partitios = userRDD.partitions();
+ System.out.println(partitios.size());
+ int[] partitionIds = new int[partitios.size()];
+ for (int i = 0; i < partitios.size(); i++) {
+ int index = partitios.get(i).index();
+ partitionIds[i] = index;
+ }
+
+ List<String>[] userIPs = userRDD.collectPartitions(partitionIds);
+ for (int i = 0; i < userIPs.length; i++) {
+ List<String> iuser = userIPs[i];
+ System.out.println(i + " partition");
+ System.out.println(iuser.toString());
+ }
+ }
+
+ public JavaRDD<String> parallizeUsers(Map<String, Double> userDocs) {
+
+ // prepare list for parallize
+ List<Tuple2<String, Double>> list = new ArrayList<>();
+ for (String user : userDocs.keySet()) {
+ list.add(new Tuple2<String, Double>(user, userDocs.get(user)));
+ }
+
+ // group users
+ ThePartitionProblemSolver solution = new KGreedyPartitionSolver();
+ Map<String, Integer> userGroups = solution.solve(userDocs, this.partition);
+
+ JavaPairRDD<String, Double> pairRdd = spark.sc.parallelizePairs(list);
+ JavaPairRDD<String, Double> userPairRDD = pairRdd.partitionBy(new logPartitioner(userGroups, this.partition));
+
+ // repartitioned user RDD
+ return userPairRDD.keys();
+ }
+
+ public Terms getSessionTerms() {
+
+ int docCount = es.getDocCount(this.logIndex, this.cleanupType);
+
+ SearchResponse sr = es.getClient().prepareSearch(this.logIndex).setTypes(this.cleanupType).setQuery(QueryBuilders.matchAllQuery())
+ .addAggregation(AggregationBuilders.terms("Sessions").field("SessionID").size(docCount)).execute().actionGet();
+
+ Terms Sessions = sr.getAggregations().get("Sessions");
+ return Sessions;
+ }
+
+ public List<String> getSessions() {
+
+ Terms sessions = this.getSessionTerms();
+ List<String> sessionList = new ArrayList<>();
+ for (Terms.Bucket entry : sessions.getBuckets()) {
+ if (entry.getDocCount() >= 3 && !entry.getKey().equals("invalid")) {
+ String session = (String) entry.getKey();
+ sessionList.add(session);
+ }
+ }
+
+ return sessionList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sdap-mudrod/blob/39379fa9/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/RankingTrainDataGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/RankingTrainDataGenerator.java b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/RankingTrainDataGenerator.java
new file mode 100644
index 0000000..766e853
--- /dev/null
+++ b/core/src/main/java/org/apache/sdap/mudrod/weblog/pre/RankingTrainDataGenerator.java
@@ -0,0 +1,54 @@
+package org.apache.sdap.mudrod.weblog.pre;
+
+import org.apache.sdap.mudrod.discoveryengine.DiscoveryStepAbstract;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.weblog.structure.RankingTrainData;
+import org.apache.sdap.mudrod.weblog.structure.SessionExtractor;
+import org.apache.spark.api.java.JavaRDD;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+public class RankingTrainDataGenerator extends DiscoveryStepAbstract {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(RankingTrainDataGenerator.class);
+
+ public RankingTrainDataGenerator(Properties props, ESDriver es, SparkDriver spark) {
+ super(props, es, spark);
+ // TODO Auto-generated constructor stub
+ }
+
+ @Override
+ public Object execute() {
+ // TODO Auto-generated method stub
+ LOG.info("Starting generate ranking train data.");
+ startTime = System.currentTimeMillis();
+
+ String rankingTrainFile = "E:\\Mudrod_input_data\\Testing_Data_4_1monthLog+Meta+Onto\\traing.txt";
+ try {
+ SessionExtractor extractor = new SessionExtractor();
+ JavaRDD<RankingTrainData> rankingTrainDataRDD = extractor.extractRankingTrainData(this.props, this.es, this.spark);
+
+ JavaRDD<String> rankingTrainData_JsonRDD = rankingTrainDataRDD.map(f -> f.toJson());
+
+ rankingTrainData_JsonRDD.coalesce(1, true).saveAsTextFile(rankingTrainFile);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ endTime = System.currentTimeMillis();
+ LOG.info("Ranking train data generation complete. Time elapsed {} seconds.", (endTime - startTime) / 1000);
+ return null;
+ }
+
+ @Override
+ public Object execute(Object o) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+}