update rtdmp add lazada

src/main/scala/mobvista/dmp/datasource/rtdmp/lazada/Constant.scala

@@ -28,13 +28,21 @@ object Constant {
 
   val merge_install_sql: String =
     """
-      |SELECT campaign_id, gaid, '@new_date' update_date
-      |  FROM dwh.etl_adn_tracking_install
-      |  WHERE dt = '@dt'
-      |  UNION ALL
-      |SELECT campaign_id, gaid, update_date
-      |  FROM dwh.merge_adn_tracking_install
-      |  WHERE dt = '@dt' AND update_date > '@update_date'
+      |SELECT
+      |  COALESCE(t1.campaign_id, t2.campaign_id) campaign_id,
+      |  COALESCE(t1.gaid, t2.gaid) gaid,
+      |  COALESCE(t1.update_date, t2.update_date) update_date
+      |  FROM
+      |  (SELECT campaign_id, gaid, '@new_date' update_date
+      |     FROM dwh.etl_adn_tracking_install
+      |     WHERE dt = '@dt'
+      |  ) t1
+      |  FULL OUTER JOIN
+      |  (SELECT campaign_id, gaid, update_date
+      |     FROM dwh.merge_adn_tracking_install
+      |     WHERE dt = '@dt' AND update_date > '@update_date'
+      |  ) t2
+      |  ON t1.campaign_id = t2.campaign_id AND t1.gaid = t2.gaid
       |""".stripMargin
 
   val process_rtdmp_audience_sql: String =

src/main/scala/mobvista/dmp/datasource/rtdmp/lazada/MergeInstallJob.scala

@@ -37,10 +37,13 @@ class MergeInstallJob extends CommonSparkJob with Serializable {
 
       val update_date = DateUtil.getDay(DateUtil.parse(dt, "yyyyMMdd"), "yyyy-MM-dd", -30)
 
+      val before_date = DateUtil.getDay(DateUtil.parse(dt, "yyyyMMdd"), "yyyyMMdd", -1)
+
       val sql: String = Constant.merge_install_sql
         .replace("@dt", dt)
         .replace("@new_date", new_date)
         .replace("@update_date", update_date)
+        .replace("@before_date", before_date)
 
       spark.sql(sql)
         .coalesce(100)