update lazada_rtdmp,fix RTDmpFetch

src/main/java/mobvista/dmp/datasource/rtdmp/RTDmpFetch.java

@@ -121,11 +121,11 @@ public class RTDmpFetch {
         LOGGER.info("checkRules -->> audienceId:" + audienceId + ", jsonObject:" + jsonObject + ", startTime:" + startTime + ", endTime:" + endTime);
         Tuple tuple = checkRules(jsonObject, startTime, endTime);
         if (tuple.getFlag()) {
-            KV kv = mySqlUtil.getPartitionTime("dwh", " audience_merge_v1");
+            KV kv = mySqlUtil.getPartitionTime("dwh", "audience_merge_v1");
             //  retry getPartitionTime
             for (int i = 0; i < 5; i++) {
                 if (StringUtils.isBlank(kv.getK())) {
-                    kv = mySqlUtil.getPartitionTime("dwh", " audience_merge_v1");
+                    kv = mySqlUtil.getPartitionTime("dwh", "audience_merge_v1");
                 } else {
                     break;
                 }

src/main/java/mobvista/dmp/util/MySQLUtil.java

@@ -78,8 +78,11 @@ public class MySQLUtil {
             stmt = conn.createStatement();
             String sql;
             //  last partition
-            sql = "SELECT part,utime FROM table_info WHERE db_name = '" + dbName + "' AND tb_name = '" + tbName + "' AND flag = 1 ORDER BY part DESC LIMIT 1";
+            sql = "SELECT part,utime FROM dmp.table_info WHERE db_name = '" + dbName + "' AND tb_name = '" + tbName + "' AND flag = 1 ORDER BY part DESC LIMIT 1";
             ResultSet rs = stmt.executeQuery(sql);
+            if (!rs.next()) {
+                System.out.println("getPartitionTime Null,SQL --->>> " + sql);
+            }
             while (rs.next()) {
                 partition = rs.getString("part");
                 utime = rs.getTimestamp("utime");

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

@@ -4,8 +4,10 @@ import com.alibaba.fastjson.{JSONArray, JSONObject}
 import mobvista.dmp.common.{CommonSparkJob, MobvistaConstant}
 import mobvista.dmp.datasource.rtdmp.{Logic, ServerUtil}
 import org.apache.commons.cli.{BasicParser, Options}
+import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.io.compress.GzipCodec
 
+import java.net.URI
 import scala.collection.mutable
 
 /**
@@ -42,6 +44,8 @@ class ProcessRTJob extends CommonSparkJob with Serializable {
       val sql: String = Constant.process_rtdmp_audience_sql
         .replace("@dt", dt)
 
+      FileSystem.get(new URI(s"s3://mob-emr-test"), sc.hadoopConfiguration).delete(new Path(output), true)
+
       spark.sql(sql)
         .rdd.map(row => {
         row.getAs[String]("gaid")