Merge pull request #16 from Java-Cesco/calcDelta

Calc delta

@@ -9,8 +9,5 @@
         <module name="Detecting_fraud_clicks" />
       </profile>
     </annotationProcessing>
-    <bytecodeTargetLevel target="9">
-      <module name="Detecting_fraud_clicks" target="1.8" />
-    </bytecodeTargetLevel>
   </component>
 </project>
\ No newline at end of file

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="MarkdownExportedFiles">
+    <htmlFiles />
+    <imageFiles />
+    <otherFiles />
+  </component>
+</project>
\ No newline at end of file

 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="JavaScriptSettings">
-    <option name="languageLevel" value="ES6" />
+  <component name="ExternalStorageConfigurationManager" enabled="true" />
+  <component name="MavenProjectsManager">
+    <option name="originalFiles">
+      <list>
+        <option value="$PROJECT_DIR$/pom.xml" />
+      </list>
+    </option>
+  </component>
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" project-jdk-name="1.8" project-jdk-type="JavaSDK">
+    <output url="file://$PROJECT_DIR$/out" />
   </component>
   <component name="MavenProjectsManager">
     <option name="originalFiles">

-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="ProjectModuleManager">
-    <modules>
-      <module fileurl="file://$PROJECT_DIR$/.idea/Detecting_fraud_clicks.iml" filepath="$PROJECT_DIR$/.idea/Detecting_fraud_clicks.iml" />
-    </modules>
-  </component>
-</project>
\ No newline at end of file

 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="VcsDirectoryMappings">
-    <mapping directory="" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
   </component>
 </project>
\ No newline at end of file

-import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
-import scala.Tuple2;
-
-import java.util.Arrays;
-import java.util.List;
-
-public class MapExample {
-
-    static SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Cesco");
-    static JavaSparkContext sc = new JavaSparkContext(conf);
-    
-    public static void main(String[] args) throws Exception {
-        
-        // Parallelized with 2 partitions
-        JavaRDD<String> x = sc.parallelize(
-                Arrays.asList("spark", "rdd", "example", "sample", "example"),
-                2);
-
-        // Word Count Map Example
-        JavaRDD<Tuple2<String, Integer>> y1 = x.map(e -> new Tuple2<>(e, 1));
-        List<Tuple2<String, Integer>> list1 = y1.collect();
-
-        // Another example of making tuple with string and it's length
-        JavaRDD<Tuple2<String, Integer>> y2 = x.map(e -> new Tuple2<>(e, e.length()));
-        List<Tuple2<String, Integer>> list2 = y2.collect();
-        
-        System.out.println(list1);
-    }
-}

+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.expressions.Window;
+import org.apache.spark.sql.expressions.WindowSpec;
+
+import javax.xml.crypto.Data;
+
+import static org.apache.spark.sql.functions.*;
+
+public class calForwardTimeDelta {
+ static SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Cesco");
+ static JavaSparkContext sc = new JavaSparkContext(conf);
+
+ public static void main(String[] args) throws Exception{
+     //Create Session
+     SparkSession spark = SparkSession
+             .builder()
+             .appName("Detecting Fraud Clicks")
+             .getOrCreate();
+
+     //run methods hereu
+     calcDelta(spark);
+ }
+
+ private static void calcDelta(SparkSession spark){
+     // put the path the file you gonna deal with being placed
+     String filepath =  "train_sample.csv";
+
+     // create Dataset from files
+     Dataset<Row> logDF = spark.read()
+             .format("csv")
+             .option("inferSchema", "true")
+             .option("header","true")
+             .load(filepath);
+
+     // cast timestamp(click_time, attributed_time) type to long type
+
+     //add column for long(click_time)
+     Dataset<Row> newDF = logDF.withColumn("utc_click_time", logDF.col("click_time").cast("long"));
+     //add column for long(attributed_time)
+     newDF = newDF.withColumn("utc_attributed_time", logDF.col("attributed_time").cast("long"));
+     //drop timestamp type columns
+     newDF = newDF.drop("click_time").drop("attributed_time");
+     newDF.createOrReplaceTempView("logs");
+
+     WindowSpec w = Window.partitionBy ("ip")
+             .orderBy("utc_click_time");
+
+     newDF = newDF.withColumn("lag(utc_click_time)", lag("utc_click_time",1).over(w));
+     newDF.where("ip=10").show();
+     newDF = newDF.withColumn("delta", when(col("lag(utc_click_time)").isNull(),lit(0)).otherwise(col("utc_click_time")).minus(when(col("lag(utc_click_time)").isNull(),lit(0)).otherwise(col("lag(utc_click_time)"))));
+     //newDF = newDF.withColumn("delta", datediff());
+     newDF = newDF.drop("lag(utc_click_time)");
+     newDF = newDF.orderBy("ip");
+
+     newDF.show();
+ }
+
+}

-public class valid {
-    private int x;
-    
-    valid() {
-        x = 0;
-    }
-    
-    void printX(){
-        System.out.println(x);
-    }
-    
-    public static void main(String[] args){
-        valid v = new valid();
-        v.printX();
-    }
-    
-}