Commit eb25d60633281ed71962a572e490e6e777b7b01a
1 parent
783fe3a33d
Exists in
master
editor score
Showing 1 changed file with 4 additions and 2 deletions Side-by-side Diff
app/com/piki_ds/ver1/EditorScore.scala
View file @
eb25d60
| ... | ... | @@ -38,7 +38,7 @@ |
| 38 | 38 | list.filter(x=>x.isDirectory && (!isParted || (isParted && hdfs.exists(x.getPath.suffix("/_SUCCESS"))))).maxBy(x=>x.getModificationTime) |
| 39 | 39 | } |
| 40 | 40 | |
| 41 | - def followGetter(sQLContext: SQLContext, dateKey:String) = { | |
| 41 | + def followGetter(sQLContext: SQLContext, dateKey:String, fileSave:Boolean = true) = { | |
| 42 | 42 | val fromExisting = sQLContext.read.format("json").load(recentlyUpdatedPath("/preprocess/followInfo",true,fs).getPath.toString) |
| 43 | 43 | val fromUpdated = getDashTable(sQLContext, "EDITOR_FOLLOW", dateKey) |
| 44 | 44 | val unionFrom = fromExisting.unionAll(fromUpdated) |
| ... | ... | @@ -46,7 +46,9 @@ |
| 46 | 46 | Map("cdate" -> x.getAs[String](0).toString, "follow_cnt" -> x.getAs[Long](1).toString, "uid" -> x.getAs[Long](3).toString) |
| 47 | 47 | }) |
| 48 | 48 | val follow_info = intoMap.groupBy(x=>x("uid")).map(x=>(x._1, x._2.map(a=>a("follow_cnt").toLong).sum)) |
| 49 | - unionFrom.write.format("json").save(s"hdfs://pikinn/preprocess/followInfo/$dateKey") | |
| 49 | + if (fileSave) { | |
| 50 | + unionFrom.write.format("json").save(s"hdfs://pikinn/preprocess/followInfo/$dateKey") | |
| 51 | + } | |
| 50 | 52 | follow_info |
| 51 | 53 | } |
| 52 | 54 |