fixed the formatting for spark jobs

simonambridge · Apr 26, 2016 · 7aeed3c · 7aeed3c
1 parent 77c9bdc
commit 7aeed3c
Show file tree

Hide file tree

Showing 11 changed files with 99 additions and 60 deletions.
diff --git a/.DS_Store b/.DS_Store
diff --git a/RollUpReports/.idea/misc.xml b/RollUpReports/.idea/misc.xml
diff --git a/RollUpReports/.idea/workspace.xml b/RollUpReports/.idea/workspace.xml
diff --git a/RollUpReports/src/main/scala/RollUpReportsByCC.scala b/RollUpReports/src/main/scala/RollUpReportsByCC.scala
@@ -28,21 +28,24 @@ object RollUpReportsByCC {
       )""")
 
     //hourlyaggregates_bycc
-    val rollup4= sqlContext.sql("select cc_no, int(concat(year,month,day,hour)) as hour, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month,day,hour))")
+    //val rollup4= sqlContext.sql("select cc_no, int(concat(year,month,day,hour)) as hour, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month,day,hour))")
+    val rollup4= sqlContext.sql("select cc_no, int(concat(year, if(length(month)=1, concat('0',month), month),if(length(day)=1, concat('0',day), day), if(length(hour)=1, concat('0',hour), hour))) as hour, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount from temp_transactions group by cc_no, concat(year, if(length(month)=1, concat('0',month), month),if(length(day)=1, concat('0',day), day), if(length(hour)=1, concat('0',hour), hour))")
     rollup4.write.format("org.apache.spark.sql.cassandra")
       .mode(SaveMode.Overwrite)
       .options(Map("keyspace" -> "rtfap", "table" -> "hourlyaggregates_bycc"))
       .save()
 
     // dailyaggregations_bycc
-    val rollup3= sqlContext.sql("select cc_no, int(concat(year,month,day)) as day, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month,day))")
+    //val rollup3= sqlContext.sql("select cc_no, int(concat(year,month,day)) as day, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month,day))")
+    val rollup3= sqlContext.sql("select cc_no, int(concat(year, if(length(month)=1, concat('0',month), month),if(length(day)=1, concat('0',day), day))) as day, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount from temp_transactions group by cc_no, concat(year, if(length(month)=1, concat('0',month), month),if(length(day)=1, concat('0',day), day))")
     rollup3.write.format("org.apache.spark.sql.cassandra")
       .mode(SaveMode.Overwrite)
       .options(Map("keyspace" -> "rtfap", "table" -> "dailyaggregates_bycc"))
       .save()
 
     // monthlyaggregations_bycc
-    val rollup5= sqlContext.sql("select cc_no, int(concat(year,month)) as month, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month))")
+    //val rollup5= sqlContext.sql("select cc_no, int(concat(year,month)) as month, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount, count(*) as total_count from temp_transactions group by cc_no, int(concat(year,month))")
+    val rollup5= sqlContext.sql("select cc_no, int(concat(year, if(length(month)=1, concat('0',month), month))) as month, sum(amount) as total_amount, min(amount) as min_amount, max(amount) as max_amount from temp_transactions group by cc_no, concat(year, if(length(month)=1, concat('0',month), month))")
     rollup5.write.format("org.apache.spark.sql.cassandra")
       .mode(SaveMode.Overwrite)
       .options(Map("keyspace" -> "rtfap", "table" -> "monthlyaggregates_bycc"))

diff --git a/RollUpReports/target/scala-2.10/classes/RollUpReportsByCC$.class b/RollUpReports/target/scala-2.10/classes/RollUpReportsByCC$.class
diff --git a/RollUpReports/target/scala-2.10/rollupreports_2.10-1.0.jar b/RollUpReports/target/scala-2.10/rollupreports_2.10-1.0.jar
diff --git a/...s/target/streams/$global/dependencyPositions/$global/streams/update_cache_2.10/output_dsp b/...s/target/streams/$global/dependencyPositions/$global/streams/update_cache_2.10/output_dsp
diff --git a/RollUpReports/target/streams/compile/compileIncremental/$global/streams/export b/RollUpReports/target/streams/compile/compileIncremental/$global/streams/export