support rtdmp temp audience

azkaban/rtdmp/rtdmp_request.job

 type=command
-dependencies=rtdmp_request_tencent,rtdmp_request_other,rtdmp_request_dsp,rtdmp_request_btop,rtdmp_request_uc,rtdmp_request_youku_acquisition,rtdmp_request_alive
+dependencies=rtdmp_request_v2,rtdmp_request_tencent,rtdmp_request_other,rtdmp_request_dsp,rtdmp_request_btop,rtdmp_request_uc,rtdmp_request_youku_acquisition,rtdmp_request_alive
 command=echo "RTDmp Request Success !!!"
\ No newline at end of file

azkaban/rtdmp/rtdmp_request_v2.job

+type=command
+command=bash -x rtdmp_request_v2.sh
\ No newline at end of file

azkaban/rtdmp/rtdmp_request_v2.sh

+#!/bin/bash
+
+source ../dmp_env.sh
+
+today=${ScheduleTime}
+date_path=$(date +%Y/%m/%d -d "-1 day $today")
+
+INPUT_PATH="s3://mob-emr-test/dataplatform/DataWareHouse/data/dwh/tmp/rtdmp_tmp_package_name/cn_good_channel/${date_path}"
+
+audience_names="adx_goodreq7 sdk_goodreq7 sdk_wangzhuan_goodreq7"
+for audience_name in ${audience_names}; do
+  S3_PATH="${INPUT_PATH}/${audience_name}"
+  check_await ${S3_PATH}/_SUCCESS
+  java -cp ../${JAR} mobvista.dmp.datasource.rtdmp.RTDmpRequestV2 "${audience_name}" "${S3_PATH}"
+done

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

+package mobvista.dmp.datasource.rtdmp;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import mobvista.dmp.util.JdbcUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.net.URISyntaxException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.*;
+
+/**
+ * @package: mobvista.dmp.datasource.rtdmp
+ * @author: wangjf
+ * @date: 2021/8/3
+ * @time: 10:59 上午
+ * @email: jinfeng.wang@mobvista.com
+ */
+public class RTDmpRequestV2 {
+
+    public static void main(String[] args) {
+        String audience_name = "";
+        String s3_path = "";
+        if (args.length >= 2) {
+            audience_name = args[0];
+            s3_path = args[1];
+        }
+        Map<String, Integer> map = getAudienceInfo("other");
+        Set<String> uploadAudienceSet = new HashSet<>();
+
+        JSONObject jsonObject;
+        if (map.containsKey(audience_name)) {
+            jsonObject = new JSONObject();
+            jsonObject.put("id", map.get(audience_name));
+            jsonObject.put("s3_path", s3_path);
+            jsonObject.put("status", 1);
+            jsonObject.put("audience_data_status", 1);
+            JSONArray updateJsonArray = new JSONArray();
+            updateJsonArray.add(jsonObject);
+            ServerUtil.update(updateJsonArray);
+        } else {
+            jsonObject = new JSONObject();
+            jsonObject.put("s3_path", s3_path);
+            jsonObject.put("platform", 1);
+            jsonObject.put("match_device_type", "1,3,6,9");
+            jsonObject.put("audience_type", 2);
+            jsonObject.put("data_update_method", 1);
+            jsonObject.put("audience_name", audience_name);
+            jsonObject.put("status", 1);
+            jsonObject.put("audience_gender", 3);
+            jsonObject.put("audience_count", 1);
+            jsonObject.put("is_sync_dmpserver", 2);
+            jsonObject.put("is_sync_bigmedia", 1);
+            jsonObject.put("audience_data_status", 1);
+            uploadAudienceSet.add(audience_name);
+            JSONArray uploadJsonArray = new JSONArray();
+            uploadJsonArray.add(jsonObject);
+            ServerUtil.upload(uploadJsonArray);
+            writeAudienceInfo("other", getAudienceMap(uploadAudienceSet));
+        }
+
+    }
+
+    private static Map<String, Integer> getAudienceInfo(String business) {
+        Map<String, Integer> map = new java.util.HashMap<>();
+        JdbcUtils jdbcUtils = new JdbcUtils();
+        try {
+            ResultSet result = jdbcUtils.query("SELECT package_name, audience_id FROM dmp.rtdmp_audience_info WHERE business = '" + business + "'");
+            while (result.next()) {
+                String package_name = result.getString("package_name");
+                Integer audience_id = result.getInt("audience_id");
+                map.put(package_name, audience_id);
+            }
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+        return map;
+    }
+
+    private static Map<String, Integer> getAudienceMap(Set<String> pkgSet) {
+        Map<String, Integer> map = new HashMap<>();
+        for (String pkg : pkgSet) {
+            try {
+                int id = ServerUtil.query(pkg);
+                if (id > 0) {
+                    map.put(pkg, id);
+                }
+            } catch (URISyntaxException e) {
+                e.printStackTrace();
+            }
+        }
+        return map;
+    }
+
+    private static void writeAudienceInfo(String business, Map<String, Integer> map) {
+        StringJoiner values = new StringJoiner(",");
+        for (Map.Entry<String, Integer> entry : map.entrySet()) {
+            values.add("('" + business + "','" + entry.getKey() + "'," + entry.getValue() + ")");
+        }
+        if (StringUtils.isNotBlank(values.toString())) {
+            String insertSql =
+                    "REPLACE INTO dmp.rtdmp_audience_info(business, package_name, audience_id) VALUES " + values;
+            JdbcUtils jdbcUtils = new JdbcUtils();
+            try {
+                jdbcUtils.update(insertSql);
+            } catch (SQLException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}