feat: 企业变更通用程序修改

- 将handle类统一抽取
- 企业变更支持复制和非复制两种模式

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

@@ -2,12 +2,12 @@ package com.winhc.bigdata.spark.jobs.chance
 
 import com.winhc.bigdata.spark.config.EsConfig
 import com.winhc.bigdata.spark.utils.BaseUtil.isWindows
-import com.winhc.bigdata.spark.utils.{BaseUtil, LoggingUtils, SparkUtils}
+import com.winhc.bigdata.spark.utils.{BaseUtil, LoggingUtils, ReflectUtils, SparkUtils}
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.types.{MapType, StringType, StructField, StructType}
-import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.{DataFrame, Row, SparkSession}
 
 import scala.annotation.meta.getter
 import scala.collection.mutable
@@ -48,7 +48,7 @@ object ChangeExtract {
 
     val target_eci_change_extract = "ads_change_extract"
 
-    def calc(): Unit = {
+    def calc(isCopy: Boolean = true): Unit = {
       val cols = primaryFields.filter(!_.equals(primaryKey)).seq
 
       val ds = inc_ds.replace("-", "")
@@ -60,96 +60,136 @@ object ChangeExtract {
 
       val lastDs_ads_all = getLastPartitionsOrElse(s"$project.ads_$tableName", "0")
 
-      val handle = getHandleClazz(tableName, cols)
+      val handle = ReflectUtils.getClazz[CompanyChangeHandle](s"com.winhc.bigdata.spark.jobs.chance.table.$tableName", cols)
+      //      val handle = getHandleClazz(tableName, cols)
 
       val update_time = BaseUtil.nowDate()
 
-      sql(
-        s"""
-           |SELECT  cid,current_cid as new_cid
-           |FROM    ${project}.inc_ods_company
-           |WHERE   ds > $lastDs_ads_all and ds < $ds
-           |AND     cid IS NOT NULL
-           |AND     current_cid IS NOT NULL
-           |GROUP BY cid,current_cid
-           |""".stripMargin).createOrReplaceTempView("mapping")
-
-      val cid = getColumns(s"$project.ads_$tableName").filter(f => f.equals("cid") || f.equals("new_cid")).max
+      var df: DataFrame = null
+      isCopy match {
+        case true => {
+          sql(
+            s"""
+               |SELECT  cid,current_cid as new_cid
+               |FROM    ${project}.inc_ods_company
+               |WHERE   ds > $lastDs_ads_all and ds < $ds
+               |AND     cid IS NOT NULL
+               |AND     current_cid IS NOT NULL
+               |GROUP BY cid,current_cid
+               |""".stripMargin).createOrReplaceTempView("mapping")
+
+          val cid = getColumns(s"$project.ads_$tableName").filter(f => f.equals("cid") || f.equals("new_cid")).max
+          df = sql(
+            s"""
+               |SELECT  $primaryKey,${otherAllCols.mkString(",")},'0' as change_flag
+               |FROM    $project.inc_ads_$tableName
+               |WHERE   ds = $ds
+               |UNION ALL
+               |SELECT  t2.$primaryKey,${otherAllCols.map("t2." + _).mkString(",")},'1' as change_flag
+               |FROM    (
+               |            SELECT  DISTINCT ${primaryKey}
+               |            FROM    $project.inc_ads_$tableName
+               |            WHERE   ds = $ds
+               |        ) AS t1
+               |JOIN    (
+               |             SELECT  concat_ws('_',coalesce(mm.new_cid,tmp.$cid),split(rowkey, '_')[1]) AS rowkey
+               |                     ,${intersectCols.diff(Set("rowkey", "cid", "new_cid")).mkString(",")}
+               |                     ,coalesce(mm.new_cid,tmp.$cid) AS new_cid
+               |                     ,tmp.$cid as cid
+               |                     ,c
+               |             FROM    (
+               |                         SELECT  a.*
+               |                                 ,row_number() OVER (PARTITION BY a.${primaryKey} ORDER BY update_time DESC) c
+               |                         FROM    (
+               |                                     SELECT  ${intersectCols.mkString(",")}
+               |                                     FROM    $project.ads_$tableName
+               |                                     WHERE   ds = $lastDs_ads_all
+               |                                     UNION ALL
+               |                                     SELECT  ${intersectCols.mkString(",")}
+               |                                     FROM    $project.inc_ads_$tableName
+               |                                     WHERE   ds > $lastDs_ads_all and ds < $ds
+               |                                 ) AS a
+               |                     ) AS tmp
+               |             LEFT JOIN mapping mm
+               |             ON tmp.$cid = mm.cid
+               |             WHERE   tmp.c = 1
+               |        ) AS t2
+               |ON      t1.${primaryKey} = t2.${primaryKey}
+               |""".stripMargin)
+        }
+        case false => {
+          df = sql(
+            s"""
+               |SELECT  $primaryKey,${otherAllCols.mkString(",")},'0' as change_flag
+               |FROM    $project.inc_ads_$tableName
+               |WHERE   ds = $ds
+               |UNION ALL
+               |SELECT  t2.$primaryKey,${otherAllCols.map("t2." + _).mkString(",")},'1' as change_flag
+               |FROM    (
+               |            SELECT  DISTINCT ${primaryKey}
+               |            FROM    $project.inc_ads_$tableName
+               |            WHERE   ds = $ds
+               |        ) AS t1
+               |JOIN    (
+               |             SELECT  tmp.*
+               |             FROM    (
+               |                         SELECT  a.*
+               |                                 ,row_number() OVER (PARTITION BY a.${primaryKey} ORDER BY update_time DESC) c
+               |                         FROM    (
+               |                                     SELECT  ${intersectCols.mkString(",")}
+               |                                     FROM    $project.ads_$tableName
+               |                                     WHERE   ds = $lastDs_ads_all
+               |                                     UNION ALL
+               |                                     SELECT  ${intersectCols.mkString(",")}
+               |                                     FROM    $project.inc_ads_$tableName
+               |                                     WHERE   ds > $lastDs_ads_all and ds < $ds
+               |                                 ) AS a
+               |                     ) AS tmp
+               |             WHERE   tmp.c = 1
+               |        ) AS t2
+               |ON      t1.${primaryKey} = t2.${primaryKey}
+               |""".stripMargin)
+        }
+      }
 
-      val rdd = sql(
-        s"""
-           |SELECT  $primaryKey,${otherAllCols.mkString(",")},'0' as change_flag
-           |FROM    $project.inc_ads_$tableName
-           |WHERE   ds = $ds
-           |UNION ALL
-           |SELECT  t2.$primaryKey,${otherAllCols.map("t2." + _).mkString(",")},'1' as change_flag
-           |FROM    (
-           |            SELECT  DISTINCT ${primaryKey}
-           |            FROM    $project.inc_ads_$tableName
-           |            WHERE   ds = $ds
-           |        ) AS t1
-           |JOIN    (
-           |             SELECT  concat_ws('_',coalesce(mm.new_cid,tmp.$cid),split(rowkey, '_')[1]) AS rowkey
-           |                     ,${intersectCols.diff(Seq("rowkey", "cid", "new_cid")).mkString(",")}
-           |                     ,coalesce(mm.new_cid,tmp.$cid) AS new_cid
-           |                     ,tmp.$cid as cid
-           |                     ,c
-           |             FROM    (
-           |                         SELECT  a.*
-           |                                 ,row_number() OVER (PARTITION BY a.${primaryKey} ORDER BY update_time DESC) c
-           |                         FROM    (
-           |                                     SELECT  ${intersectCols.mkString(",")}
-           |                                     FROM    $project.ads_$tableName
-           |                                     WHERE   ds = $lastDs_ads_all
-           |                                     UNION ALL
-           |                                     SELECT  ${intersectCols.mkString(",")}
-           |                                     FROM    $project.inc_ads_$tableName
-           |                                     WHERE   ds > $lastDs_ads_all and ds < $ds
-           |                                 ) AS a
-           |                     ) AS tmp
-           |             LEFT JOIN mapping mm
-           |             ON tmp.$cid = mm.cid
-           |             WHERE   tmp.c = 1
-           |        ) AS t2
-           |ON      t1.${primaryKey} = t2.${primaryKey}
-           |""".stripMargin)
-        .select(all_cols.map(column => col(column).cast("string")): _*)
-        .rdd.map(r => {
-        (r.getAs[String](primaryKey), all_cols.map(f => (f, r.getAs[String](f))).toMap)
-      }).groupByKey()
-        .map(x => {
-          val rowkey = x._1
-          val map_list = x._2
-          //          try {
-          if (map_list.size == 1) {
-            val res = handle.handle(rowkey, null, map_list.head)
-            Row(res._1, res._2, tableName, res._3, res._4, res._5, res._6, res._7, res._8, update_time, res._9)
-          } else {
-            if (map_list.size > 2) {
-              logInfo("list.size greater than 2! rowkey:" + rowkey)
-            }
-            val m = getDoubleDataMap(map_list)
 
-            val new_map = m._1
-            val old_map = m._2
-            val res = handle.handle(rowkey, old_map, new_map)
-            if (res == null) {
-              null
-            } else {
+      val rdd =
+        df.select(all_cols.map(column => col(column).cast("string")): _*)
+          .rdd.map(r => {
+          (r.getAs[String](primaryKey), all_cols.map(f => (f, r.getAs[String](f))).toMap)
+        }).groupByKey()
+          .map(x => {
+            val rowkey = x._1
+            val map_list = x._2
+            //          try {
+            if (map_list.size == 1) {
+              val res = handle.handle(rowkey, null, map_list.head)
               Row(res._1, res._2, tableName, res._3, res._4, res._5, res._6, res._7, res._8, update_time, res._9)
+            } else {
+              if (map_list.size > 2) {
+                logInfo("list.size > 2! rowkey:" + rowkey)
+              }
+              val m = getDoubleDataMap(map_list)
+
+              val new_map = m._1
+              val old_map = m._2
+              val res = handle.handle(rowkey, old_map, new_map)
+              if (res == null) {
+                null
+              } else {
+                Row(res._1, res._2, tableName, res._3, res._4, res._5, res._6, res._7, res._8, update_time, res._9)
+              }
             }
-          }
-          /* } catch {
-             case e: Exception => {
-               logError(s"xjk rowkey:$rowkey msg:${e.getMessage} equCols:$cols")
-               logError(e.getMessage, e)
-               println(s"xjk rowkey:$rowkey msg:${e.getMessage} equCols:$cols")
-             }
-               null
-           }*/
-        }).filter(_ != null)
-
-      // (123_abc,insert,{a->b},all,新增某土地公示，1（1.一般变更，2.风险变更）)
+            /* } catch {
+               case e: Exception => {
+                 logError(s"xjk rowkey:$rowkey msg:${e.getMessage} equCols:$cols")
+                 logError(e.getMessage, e)
+                 println(s"xjk rowkey:$rowkey msg:${e.getMessage} equCols:$cols")
+               }
+                 null
+             }*/
+          }).filter(_ != null)
+
       val schema = StructType(Array(
         StructField("rowkey", StringType), //表数据主建
         StructField("cid", StringType), //公司id
@@ -195,10 +235,13 @@ object ChangeExtract {
   // winhc_eci_dev company_own_tax rowkey 20200729 tax_balance,tax_category,tax_num
 
 
+  //winhc_eci_dev company_equity_info id 20200730 reg_number false
+
+
   // winhc_eci_dev company cid 20200630 legal_entity_id,reg_location,business_scope,reg_status,reg_capital,emails,phones
   def main(args: Array[String]): Unit = {
-    if (args.length == 5) {
-      val Array(project, tableName, rowkey, inc_ds, pf) = args
+    if (args.length >= 5 && args.length <= 6) {
+      val Array(project, tableName, rowkey, inc_ds, pf, isCopy) = if (args.length == 6) args else args :+ "true"
       val config = EsConfig.getEsConfigMap ++ mutable.Map(
         "spark.hadoop.odps.project.name" -> project,
         "spark.hadoop.odps.spark.local.partition.amt" -> "10"
@@ -206,7 +249,7 @@ object ChangeExtract {
       val spark = SparkUtils.InitEnv("ChangeExtract", config)
 
 
-      ChangeExtractHandle(spark, project, tableName, rowkey, inc_ds, pf.split(",")).calc
+      ChangeExtractHandle(spark, project, tableName, rowkey, inc_ds, pf.split(",")).calc(isCopy.toBoolean)
       spark.stop()
     } else {
       val ds = args(0)
@@ -231,8 +274,9 @@ object ChangeExtract {
           |""".stripMargin.replace("20200717", ds)
       for (r <- rows.split("\r\n")) {
         if (StringUtils.isNotEmpty(r)) {
-          val Array(tmp, tableName, rowkey, inc_ds, pf) = r.split(" ")
-          ChangeExtractHandle(spark, project, tableName, rowkey, inc_ds, pf.split(",")).calc
+          val as = r.split(" ")
+          val Array(tmp, tableName, rowkey, inc_ds, pf, isCopy) = if (as.length == 6) as else as :+ "true"
+          ChangeExtractHandle(spark, project, tableName, rowkey, inc_ds, pf.split(",")).calc(isCopy.toBoolean)
         }
       }
       spark.stop()

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

@@ -1,7 +1,6 @@
 package com.winhc.bigdata.spark.jobs.chance
 
 import com.winhc.bigdata.spark.utils.BaseUtil.cleanup
-import com.winhc.bigdata.spark.utils.ChangeExtractUtils
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.internal.Logging
 
@@ -71,174 +70,3 @@ trait CompanyChangeHandle extends Serializable with Logging {
     }
   }
 }
-
-//土地公示
-case class company_land_publicity(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("project_name", "location", "application_name", "area", "final_price", "land_num"))
-    str
-  }
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("publication_start_date")
-
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}地块公示发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}地块公示")
-}
-
-
-case class company(equCols: Seq[String]) extends CompanyChangeHandle with Serializable {
-  override def getCid(rowkey: String, newMap: Map[String, String]): String = rowkey
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = "1"
-
-  override def getBizTime(newMap: Map[String, String]): String = "业务时间"
-
-  override def getUpdateTitle(newMap: Map[String, String]): String = "更新一家公司"
-
-  override def getInsertTitle(newMap: Map[String, String]): String = "新增一家公司"
-}
-
-//商标
-case class company_tm(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"${newMap("tm_name")}商标发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"新增${newMap("tm_name")}商标")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("商标", newMap("tm_name"), newMap("app_date"), newMap("reg_no"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
-}
-
-//专利
-case class company_patent_list(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}专利发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}专利")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("专利", newMap("title"), newMap("app_date"), newMap("app_number"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
-}
-
-//资质证书
-case class company_certificate(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"${newMap("cert_no")}资质证书发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"新增${newMap("cert_no")}资质证书")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("资质证书", newMap("cert_no"), newMap("start_date"), newMap("type"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
-}
-
-//作品著作权
-case class company_copyright_works_list(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}作品著作权发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}作品著作权")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("作品著作权", newMap("name"), newMap("reg_time"), newMap("reg_num"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
-}
-
-//软件著作权
-case class company_copyright_reg_list(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"${newMap("full_name")}软件著作权发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"新增${newMap("full_name")}软件著作权")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("软件著作权", newMap("full_name"), newMap("reg_time"), newMap("reg_num"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
-}
-
-//网站
-case class company_icp(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"${newMap("domain")}网站备案发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"新增${newMap("domain")}网站备案")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("网站备案", newMap("domain"), newMap("examine_date"), newMap("liscense"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("examine_date")
-}
-
-
-//购地信息
-case class company_land_announcement(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}购地信息发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}购地信息")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "购地信息", Array("project_name", "project_loc", "area", "tran_price", "e_number"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("contract_date")
-}
-
-//招聘
-case class company_employment(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招聘信息发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招聘信息")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招聘", Array("title", "city->employment_city", "employ_num", "start_date"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
-}
-
-//招投标
-case class company_bid_list(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招投标信息发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招投标信息")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招投标", Array("publish_time", "title", "purchaser", "province", "abs"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
-}
-
-//土地转让
-case class company_land_transfer(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"${newMap("title")}土地转让信息发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"新增${newMap("location")}土地转让信息")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "土地转让", Array("merchandise_time", "user_pre", "user_now", "location", "area", "merchandise_price", "aministrative_area"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("merchandise_time")
-}
-
-//环保处罚
-case class company_env_punishment(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"${newMap("title")}环保处罚信息发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"新增${newMap("punish_number")}环保处罚信息")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "环保处罚", Array("name", "department", "punish_number", "punish_basis", "law_break", "reason", "content", "deleted"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
-}
-
-case class company_abnormal_info(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("put_department", "remove_department", "put_reason", "put_date", "remove_date", "remove_reason"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("put_date")
-}
-
-//欠税
-case class company_own_tax(equCols: Seq[String]) extends CompanyChangeHandle {
-  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}欠税公告发生变更")
-
-  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}欠税公告")
-
-  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("欠税公告", newMap("name"), newMap("publish_date"), newMap("tax_num"))
-
-  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_date")
-}

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

@@ -0,0 +1,23 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+case class company(equCols: Seq[String]) extends CompanyChangeHandle with Serializable {
+  override def getCid(rowkey: String, newMap: Map[String, String]): String = rowkey
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = "1"
+
+  override def getBizTime(newMap: Map[String, String]): String = "业务时间"
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = "更新一家公司"
+
+  override def getInsertTitle(newMap: Map[String, String]): String = "新增一家公司"
+}

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

@@ -0,0 +1,22 @@
+
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+case class company_abnormal_info(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("put_department", "remove_department", "put_reason", "put_date", "remove_date", "remove_reason"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("put_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_bid_list.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//招投标
+
+case class company_bid_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招投标信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招投标信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招投标", Array("publish_time", "title", "purchaser", "province", "abs"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_certificate.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//资质证书
+
+case class company_certificate(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"${newMap("cert_no")}资质证书发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"新增${newMap("cert_no")}资质证书")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("资质证书", newMap("cert_no"), newMap("start_date"), newMap("type"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_copyright_reg_list.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//软件著作权
+
+case class company_copyright_reg_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"${newMap("full_name")}软件著作权发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"新增${newMap("full_name")}软件著作权")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("软件著作权", newMap("full_name"), newMap("reg_time"), newMap("reg_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_copyright_works_list.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//作品著作权
+
+case class company_copyright_works_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}作品著作权发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}作品著作权")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("作品著作权", newMap("name"), newMap("reg_time"), newMap("reg_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_employment.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//招聘
+
+case class company_employment(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招聘信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招聘信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招聘", Array("title", "city->employment_city", "employ_num", "start_date"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_env_punishment.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//环保处罚
+
+case class company_env_punishment(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"${newMap("title")}环保处罚信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"新增${newMap("punish_number")}环保处罚信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "环保处罚", Array("name", "department", "punish_number", "punish_basis", "law_break", "reason", "content", "deleted"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
+}

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

@@ -0,0 +1,17 @@
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+
+case class company_equity_info(equCols: Seq[String]) extends CompanyChangeHandle {
+
+  override def getCid(rowkey: String, newMap: Map[String, String]): String = null
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = s"股权出质信息发生变更"
+
+  override def getInsertTitle(newMap: Map[String, String]): String = s"股权出质信息发生变更"
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String  = ChangeExtractUtils.getTags(newMap, "股权出质", Array("pledgor", "pledgee", "cid", "pledgor_id", "pledgee_type", "pledgee_id", "pledgor_type"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_icp.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//网站
+
+case class company_icp(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"${newMap("domain")}网站备案发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"新增${newMap("domain")}网站备案")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("网站备案", newMap("domain"), newMap("examine_date"), newMap("liscense"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("examine_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_land_announcement.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//购地信息
+
+case class company_land_announcement(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}购地信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}购地信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "购地信息", Array("project_name", "project_loc", "area", "tran_price", "e_number"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("contract_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_land_publicity.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//土地公示
+
+case class company_land_publicity(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("project_name", "location", "application_name", "area", "final_price", "land_num"))
+    str
+  }
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publication_start_date")
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}地块公示发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}地块公示")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_land_transfer.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//土地转让
+
+case class company_land_transfer(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"${newMap("title")}土地转让信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"新增${newMap("location")}土地转让信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "土地转让", Array("merchandise_time", "user_pre", "user_now", "location", "area", "merchandise_price", "aministrative_area"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("merchandise_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_own_tax.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//欠税
+
+case class company_own_tax(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}欠税公告发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}欠税公告")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("欠税公告", newMap("name"), newMap("publish_date"), newMap("tax_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_patent_list.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//专利
+
+case class company_patent_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}专利发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}专利")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("专利", newMap("title"), newMap("app_date"), newMap("app_number"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_tm.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: XuJiakai
+ * @Date: 2020/7/9 16:44
+ * @Description:
+ */
+
+
+//商标
+
+case class company_tm(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"${newMap("tm_name")}商标发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"新增${newMap("tm_name")}商标")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("商标", newMap("tm_name"), newMap("app_date"), newMap("reg_no"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
+}