Merge remote-tracking branch 'origin/master'

# Conflicts:
#	src/main/scala/com/winhc/bigdata/spark/jobs/CompanyForCid.scala
#	src/main/scala/com/winhc/bigdata/spark/jobs/CompanyIncrForCid.scala

src/main/scala/com/winhc/bigdata/spark/jobs/CalcIncrTotal.scala

@@ -42,6 +42,12 @@ object CalcIncrTotal {
   //  winhc_eci_dev company_double_random_check_result_info main_id,check_item,new_cid cid
 
   //  winhc_eci_dev company_judicial_sale_combine main_id,new_cid cids
+  //  winhc_eci_dev company_zxr_final_case new_cid,case_no cid
+
+  //  winhc_eci_dev company_license new_cid,license_number cid
+  //  winhc_eci_dev company_license_creditchina new_cid,licence_number cid
+  //  winhc_eci_dev company_license_entpub new_cid,license_number cid
+
 
   def main(args: Array[String]): Unit = {
 

src/main/scala/com/winhc/bigdata/spark/jobs/CompanyForCid.scala

@@ -26,6 +26,11 @@ object CompanyForCid {
 //  winhc_eci_dev company_tax_contravention taxpayer_number,case_info,new_cid
 //  winhc_eci_dev company_double_random_check_info check_task_num,new_cid
 //  winhc_eci_dev company_double_random_check_result_info main_id,check_item,new_cid
+//  winhc_eci_dev company_zxr_final_case case_no,new_cid
+
+//  winhc_eci_dev company_license license_number,new_cid
+//  winhc_eci_dev company_license_creditchina licence_number,new_cid
+//  winhc_eci_dev company_license_entpub license_number,new_cid
 //  winhc_eci_dev ods_company_finance new_cid,round,money
 //  winhc_eci_dev ods_company_dishonest_info new_cid,case_no
 

src/main/scala/com/winhc/bigdata/spark/jobs/CompanyIncrForCid.scala

@@ -16,6 +16,10 @@ object CompanyIncrForCid {
   //winhc_eci_dev company_app_info icon_oss_path,brief,name,new_cid
   //winhc_eci_dev ads_company_tm app_date,tm_name,reg_no,new_cid
   //winhc_eci_dev company_wechat title,public_num,new_cid
+
+  //  winhc_eci_dev company_license license_number,new_cid
+  //  winhc_eci_dev company_license_creditchina licence_number,new_cid
+  //  winhc_eci_dev company_license_entpub license_number,new_cid
   // winhc_eci_dev company_change new_cid,change_item,change_time
   // winhc_eci_dev company_illegal_info new_cid,put_reason,put_date,put_department
   //  winhc_eci_dev company_finance new_cid,round,money

src/main/scala/com/winhc/bigdata/spark/jobs/chance/ChangeExtract.scala

@@ -310,6 +310,13 @@ object ChangeExtract {
     , Args(tableName = "company_judicial_sale_combine_list", primaryFields = "title")//司法拍卖
     , Args(tableName = "company_tax_contravention", primaryFields = "case_type")//税收违法
     , Args(tableName = "company_send_announcement_list", primaryFields = "title")//送达公告
+    , Args(tableName = "company_annual_report_out_guarantee", primaryFields = "id")//年报-对外担保
+    , Args(tableName = "company_zxr_restrict", primaryFields = "status")//限制消费令，发现最新状态
+
+    , Args(tableName = "company_zxr_final_case", primaryFields = "identity_num")//终本案件
+    , Args(tableName = "company_license_creditchina", primaryFields = "licence_content")//行政许可-信用中国
+    , Args(tableName = "company_license_entpub", primaryFields = "license_name")//行政许可-企业公示
+    , Args(tableName = "company_license", primaryFields = "license_name")//行政许可
 
     , Args(tableName = "company_certificate", primaryFields = "type")
     , Args(tableName = "company_abnormal_info", primaryFields = "remove_reason")

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_annual_report_out_guarantee.scala

@@ -0,0 +1,24 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+
+/**
+ * @Author: yyn
+ * @Date: 2020/8/13
+ * @Description:
+ */
+
+
+//年报-对外担保
+
+case class company_annual_report_out_guarantee(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = "对外担保发生变更"
+
+  override def getInsertTitle(newMap: Map[String, String]): String = "新增对外担保"
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "对外担保", Array("credito_type", "credito_amount", "credito_term", "guarantee_term", "guarantee_way"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("update_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_license.scala

@@ -0,0 +1,28 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11
+ * @Description:行政许可
+ */
+
+case class company_license(equCols: Seq[String]) extends CompanyChangeHandle with Serializable  {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("license_name"), s"行政许可发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("license_name"), s"新增行政许可")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "行政许可", Array("license_name","start_date"))
+
+  override def getBizTime(newMap: Map[String, String]): String = {
+    if(StringUtils.isBlank(newMap("start_date"))){
+      newMap("update_time")
+    }else{
+      newMap("start_date").replaceAll("年","-").replaceAll("月","-").replaceAll("日","-")
+    }
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_license_creditchina.scala

@@ -0,0 +1,28 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11
+ * @Description:行政许可-信用中国
+ */
+
+case class company_license_creditchina(equCols: Seq[String]) extends CompanyChangeHandle with Serializable  {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("licence_number"), s"行政许可-信用中国发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("licence_number"), s"新增行政许可-信用中国")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "行政许可-信用中国", Array("licence_content","decision_date"))
+
+  override def getBizTime(newMap: Map[String, String]): String = {
+    if(StringUtils.isBlank(newMap("decision_date"))){
+      newMap("update_time")
+    }else{
+      newMap("decision_date")
+    }
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_license_entpub.scala

@@ -0,0 +1,28 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11
+ * @Description:行政许可-企业公示
+ */
+
+case class company_license_entpub(equCols: Seq[String]) extends CompanyChangeHandle with Serializable  {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("license_name"), s"行政许可-企业公示发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("license_name"), s"新增行政许可-企业公示")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "行政许可-企业公示", Array("license_name","start_date"))
+
+  override def getBizTime(newMap: Map[String, String]): String = {
+    if(StringUtils.isBlank(newMap("start_date"))){
+      newMap("update_time")
+    }else{
+      newMap("start_date").replaceAll("年","-").replaceAll("月","-").replaceAll("日","-")
+    }
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_zxr_final_case.scala

@@ -0,0 +1,28 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11
+ * @Description:终本案件
+ */
+
+case class company_zxr_final_case(equCols: Seq[String]) extends CompanyChangeHandle with Serializable  {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"终本案件发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增终本案件")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "终本案件", Array("identity_num","case_create_time"))
+
+  override def getBizTime(newMap: Map[String, String]): String = {
+    if(StringUtils.isBlank(newMap("case_create_time"))){
+      newMap("update_time")
+    }else{
+      newMap("case_create_time")
+    }
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_zxr_restrict.scala

@@ -0,0 +1,28 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+
+/**
+ * @Author: Yan Yongnian
+ * @Date: 2020/8/14
+ * @Description:
+ */
+
+
+//限制消费令
+
+case class company_zxr_restrict(equCols: Seq[String]) extends CompanyChangeHandle with Serializable {
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = {
+    val str = ChangeExtractUtils.getTags(newMap, "被限制高消费", Array("name", "identity_num", "court_name", "case_create_time", "case_no"))
+    str
+  }
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("case_create_time")
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("case_no"), s"${newMap("title")}限制消费令发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("case_no"), s"新增${newMap("title")}限制消费令")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/CompanyDynamic.scala

@@ -198,6 +198,20 @@ object CompanyDynamic {
     , Args(tableName = "company_finance", bName = 1)
     , Args(tableName = "company_dishonest_info", bName = 1)
     , Args(tableName = "company_send_announcement_list", bName = 1)
+    , Args(tableName = "company_annual_report_out_guarantee", bName = 1)
+    , Args(tableName = "company_zxr_restrict", bName = 1)
+
+    , Args(tableName = "company_zxr_final_case", bName = 1)//终本案件
+    , Args(tableName = "company_license_creditchina", bName = 1)//行政许可-信用中国
+    , Args(tableName = "company_license_entpub", bName = 1)//行政许可-企业公示
+    , Args(tableName = "company_license", bName = 1)//行政许可
+    , Args(tableName = "company_check_info", bName = 1)//抽查检查
+    , Args(tableName = "company_court_announcement_list", bName = 1)//法院公告
+    , Args(tableName = "company_court_open_announcement_list", bName = 1)//开庭公告
+    , Args(tableName = "company_court_register_list", bName = 1)//立案信息
+    , Args(tableName = "company_double_random_check_info", bName = 1)//双随机抽查
+    , Args(tableName = "company_judicial_sale_combine_list", bName = 1)//司法拍卖
+    , Args(tableName = "company_tax_contravention", bName = 1)//税收违法
   )
 
   private case class Args(project: String = "winhc_eci_dev"

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/CompanyDynamicHandle.scala

@@ -45,12 +45,18 @@ trait CompanyDynamicHandle {
     , "bankruptcy_open_case" -> "bankruptcy_open_case" //破产重整
     , "company_stock_announcement" -> "company_stock_announcement" //企业公告
     , "company_send_announcement_list" -> "company_send_announcement_list" //送达公告
+    , "company_annual_report_out_guarantee" -> "company_annual_report_out_guarantee" //年报-对外担保
+    , "company_zxr_restrict" -> "company_zxr_restrict" //年报-对外担保
 
     , "company_staff" -> "company_staff" //主要成员
     , "company_finance" -> "company_finance" //融资动态
     , "company_check_info" -> "spot_check" //抽查检查
     , "company_double_random_check_info" -> "company_double_random_check_info" //双随机抽查
     , "company_court_register" -> "company_court_register" //立案信息
+    , "company_zxr_final_case" -> "company_zxr_final_case" //终本案件
+    , "company_license_creditchina" -> "company_license_creditchina" //行政许可-信用中国
+    , "company_license_entpub" -> "company_license_entpub" //行政许可-企业公示
+    , "company_license" -> "company_license" //行政许可
   )
 
   private val table_2_info_type = Map(
@@ -92,13 +98,17 @@ trait CompanyDynamicHandle {
     , "company_finance" -> "31" // 融资动态
     , "company_stock_announcement" -> "32" // 企业公告
     , "company_check_info" -> "33" // 抽查检查
-    , "" -> "34" // 行政许可
+    , "company_license" -> "34-1" // 行政许可
+    , "company_license_creditchina" -> "34-2" // 行政许可-信用中国
+    , "company_license_entpub" -> "34-3" // 行政许可-企业公示
     , "company_double_random_check_info" -> "35" // 双随机抽查
-    , "" -> "36" // 限制高消费
+    , "company_zxr_restrict" -> "36" // 限制高消费
     , "" -> "37" // 被执行人
     , "company_send_announcement_list" -> "38" // 送达报告
     , "bankruptcy_open_case" -> "39" // 破产重整
     , "company_court_register" -> "40" // 立案信息
+    , "company_annual_report_out_guarantee" -> "41" // 年报-对外担保
+    , "company_zxr_final_case" -> "41" // 终本案件
   )
 
 

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_annual_report_out_guarantee.scala

@@ -0,0 +1,46 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author yyn
+ * @Date 2020/8/13
+ * @Description TODO
+ */
+//年报-对外担保
+case class company_annual_report_out_guarantee()extends CompanyDynamicHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = "对外担保"
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String = null): String = new_map("credito_term") + new_map("guarantee_way")
+
+  /**
+   * 变更时间
+   *
+   * @param new_map
+   * @return
+   */
+//  override def get_change_time(new_map: Map[String, String]): String = new_map("biz_date")
+
+  /**
+   * 风险等级
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "提示信息"
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_license.scala

@@ -0,0 +1,48 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11 17:39
+ * @Description: 行政许可
+ */
+case class company_license() extends CompanyDynamicHandle {
+
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = Seq("insert")
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+
+    s"""
+       |许可证编号：${new_map("license_number")}\n
+       |有效期自：${new_map("start_date")}\n
+       |有效期至：${new_map("end_date")}\n
+       |许可机关：${new_map("department")}\n
+       |许可内容：${new_map("scope")}
+       |""".stripMargin
+  }
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String): String = ""
+
+  override protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "提示信息"
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_license_creditchina.scala

@@ -0,0 +1,49 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11 17:39
+ * @Description: 行政许可-信用中国
+ */
+case class company_license_creditchina() extends CompanyDynamicHandle {
+
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = Seq("insert")
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+
+    s"""
+       |许可证编号：${new_map("licence_number")}\n
+       |有效期自：${new_map("decision_date")}\n
+       |有效期至：${new_map("end_date")}\n
+       |许可机关：${new_map("department")}\n
+       |许可内容：${new_map("resume")}
+       |""".stripMargin
+  }
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String): String = ""
+
+  override protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "提示信息"
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_license_entpub.scala

@@ -0,0 +1,48 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11 17:39
+ * @Description: 行政许可-企业公示
+ */
+case class company_license_entpub() extends CompanyDynamicHandle {
+
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = Seq("insert")
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+
+    s"""
+       |许可证编号：${new_map("license_number")}\n
+       |有效期自：${new_map("start_date")}\n
+       |有效期至：${new_map("end_date")}\n
+       |许可机关：${new_map("department")}\n
+       |许可内容：${new_map("scope")}
+       |""".stripMargin
+  }
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String): String = ""
+
+  override protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "提示信息"
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_zxr_final_case.scala

@@ -0,0 +1,58 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author: π
+ * @Date: 2020/8/11 17:39
+ * @Description: 终本案件
+ */
+case class company_zxr_final_case() extends CompanyDynamicHandle {
+
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = Seq("insert")
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+
+    s"""
+       |案号：${new_map("case_no")}\n
+       |执行法院：${new_map("court_name")}\n
+       |立案日期：${new_map("case_create_time")}\n
+       |终本日期：${new_map("case_final_time")}
+       |""".stripMargin
+  }
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String): String = new_map.toJson(Seq(
+    "name->被执行人姓名"
+    ,"identity_num->身份证号码/组织机构代码"
+    ,"sex->性别"
+    ,"case_no->案号"
+    ,"court_name->执行法院"
+    ,"case_create_time->立案时间"
+    ,"case_final_time->终本日期"
+    ,"exec_amount->执行标的（元）"
+    ,"no_exec_amount->未履行金额"
+  ))
+
+  override protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "风险信息"
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/tables/company_zxr_restrict.scala

@@ -0,0 +1,51 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+/**
+ * @Author yyn
+ * @Date 2020/8/14
+ * @Description TODO
+ */
+//限制消费令
+case class company_zxr_restrict()extends CompanyDynamicHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = new_map("case_no")
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String = null): String = {
+    s"""案号:$new_map("case_no")\n
+       |限消令对象：$new_map("name")\n
+       |立案日期：$new_map("case_create_time")\n
+       |发布日期：$new_map("update_time")\n""".stripMargin
+  }
+
+  /**
+   * 变更时间
+   *
+   * @param new_map
+   * @return
+   */
+//  override def get_change_time(new_map: Map[String, String]): String = new_map("biz_date")
+
+  /**
+   * 风险等级
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "高风险"
+}

src/main/scala/com/winhc/bigdata/spark/utils/CompanyIncrForCidWithoutMD5Utils.scala

@@ -17,7 +17,8 @@ import scala.collection.mutable
 case class CompanyIncrForCidWithoutMD5Utils(s: SparkSession,
                                             project: String, //表所在工程名
                                             tableName: String, //表名（不加前后辍）
-                                            dupliCols: Seq[String] // 去重列
+                                            dupliCols: Seq[String], // 去重列
+                                            updateCol: String = "update_time" //ROW_NUMBER窗口函数的ORDER BY字段，默认（可以不传参数）为update_time
                                  ) extends LoggingUtils with CompanyMapping{
   @(transient@getter) val spark: SparkSession = s
 
@@ -100,7 +101,7 @@ case class CompanyIncrForCidWithoutMD5Utils(s: SparkSession,
          |                    ,new_cid
          |                    ,cid
          |                    ,${columns.mkString(",")}
-         |                    ,ROW_NUMBER() OVER (PARTITION BY cleanup(CONCAT_WS('',${dupliCols.mkString(",")})) ORDER BY update_time DESC ) num
+         |                    ,ROW_NUMBER() OVER (PARTITION BY cleanup(CONCAT_WS('',${dupliCols.mkString(",")})) ORDER BY NVL($updateCol,update_time) DESC ) num
          |            FROM    (
          |                        SELECT  "0" AS flag
          |                                ,a.new_cid
@@ -154,7 +155,7 @@ object CompanyIncrForCidWithoutMD5Utils {
 
   def main(args: Array[String]): Unit = {
 
-    val Array(project, tableName, dupliCols, flag) = args
+    val Array(project, tableName, dupliCols, flag, _) = args
     println(
       s"""
          |project: $project
@@ -162,17 +163,22 @@ object CompanyIncrForCidWithoutMD5Utils {
          |dupliCols: $dupliCols
          |flag: $flag
          |""".stripMargin)
-    if (args.length != 4) {
-      println("请输入 project:项目, tableName:表名, dupliCols:去重字段, flag:标识 !!!")
+    if (args.length < 4) {
+      println("请输入 project:项目, tableName:表名, dupliCols:去重字段, flag:标识, [updateCol:排序列]!!!")
       sys.exit(-1)
     }
+    //ROW_NUMBER窗口函数的ORDER BY字段，默认（可以不传参数）为update_time
+    var updateCol: String = "update_time"
+    if(args.length == 5){
+      updateCol = args(4)
+    }
     val config = mutable.Map(
       "spark.hadoop.odps.project.name" -> "winhc_eci_dev",
       "spark.hadoop.odps.spark.local.partition.amt" -> "1"
     )
     val spark: SparkSession = SparkUtils.InitEnv(this.getClass.getSimpleName, config)
     flag match {
-      case "cid" => CompanyIncrForCidWithoutMD5Utils(spark, project, tableName, (dupliCols.split(",").toSeq)).calc()
+      case "cid" => CompanyIncrForCidWithoutMD5Utils(spark, project, tableName, (dupliCols.split(",").toSeq), updateCol).calc()
     }
     spark.stop()
   }