Skip to content

M
mobvista-dmp
Commit afb611c4 by wang-jinfeng
Options

fix RTDmpMainPre

parent ba33fc72
Showing with 3 additions and 1 deletion
src/main/scala/mobvista/dmp/datasource/rtdmp/RTDmpMainPre.scala
...@@ -130,7 +130,7 @@ class RTDmpMainPre extends CommonSparkJob with Serializable { ...@@ -130,7 +130,7 @@ class RTDmpMainPre extends CommonSparkJob with Serializable {
}).union(newAudience) // 与最新的人群包进行合并操作 }).union(newAudience) // 与最新的人群包进行合并操作
*/ */
newAudience newAudience
} else { } else if (dmap.size == 1) {
val audData = dmap.toList.sortWith(_._2 > _._2).take(1) val audData = dmap.toList.sortWith(_._2 > _._2).take(1)
val pathUri = new URI(audData.get(0)._1) val pathUri = new URI(audData.get(0)._1)
if (audData.nonEmpty && FileSystem.get(new URI(s"${pathUri.getScheme}://${pathUri.getHost}"), sc.hadoopConfiguration) if (audData.nonEmpty && FileSystem.get(new URI(s"${pathUri.getScheme}://${pathUri.getHost}"), sc.hadoopConfiguration)
...@@ -151,6 +151,8 @@ class RTDmpMainPre extends CommonSparkJob with Serializable { ...@@ -151,6 +151,8 @@ class RTDmpMainPre extends CommonSparkJob with Serializable {
} else { // 如果没有，则创建 空RDD } else { // 如果没有，则创建 空RDD
sc.emptyRDD[(String, Int)] sc.emptyRDD[(String, Int)]
} }
} else {
sc.emptyRDD[(String, Int)]
} }
// 所有人群包进行合并操作 // 所有人群包进行合并操作
mergeRDD = mergeRDD.union(updateRDD) mergeRDD = mergeRDD.union(updateRDD)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment