增加id监控

src/main/scala/com/winhc/bigdata/spark/ng/change/NgChangeExtract.scala

@@ -177,9 +177,9 @@ object NgChangeExtract {
 
 
   private val startArgs = Seq(
-    Args(tableName = "company_holder", primaryFields = "percent,amount,deleted")
+    Args(tableName = "company_holder", primaryFields = "holder_id,percent,amount,deleted")
     , Args(tableName = "company_staff", primaryFields = "staff_type,deleted")
-    , Args(tableName = "company", primaryKey = "company_id", primaryFields = "name,cate_third_code,county_code,reg_capital_amount,legal_entity_name,deleted")
+    , Args(tableName = "company", primaryKey = "company_id", primaryFields = "name,cate_third_code,county_code,reg_capital_amount,legal_entity_name,legal_entity_id,deleted")
     , Args(tableName = "company_tm", primaryFields = "status")
     , Args(tableName = "company_icp", primaryFields = "domain")
   )

src/main/scala/com/winhc/bigdata/spark/ng/relation/inc_company_relation_v2.scala

@@ -105,7 +105,7 @@ case class inc_company_relation_v2(s: SparkSession,
          |        ,new_data['deleted'] AS deleted
          |FROM    $ads_change_extract
          |WHERE   ds = '$ds' AND     tn = 'company'
-         |AND     (change_fields LIKE '%legal_entity_name%' or change_fields is null or change_fields LIKE '%deleted%')
+         |AND     (change_fields LIKE '%legal_entity_name%' or change_fields is null or change_fields LIKE '%deleted%' OR change_fields LIKE '%legal_entity_id%')
          |AND     length(trim(new_data['legal_entity_name'])) > 0
          |UNION ALL
          |SELECT  CONCAT_WS('_',company_id,hash(old_data['legal_entity_name'])) AS rowkey
@@ -357,7 +357,7 @@ case class inc_company_relation_v2(s: SparkSession,
          |    SELECT rowkey
          |    FROM $ads_change_extract
          |    WHERE   ds = '$ds' AND tn = 'company_holder'
-         |    AND    (change_fields LIKE '%percent%' OR change_fields is null OR change_fields LIKE '%deleted%')
+         |    AND    (change_fields LIKE '%percent%' OR change_fields is null OR change_fields LIKE '%deleted%' OR change_fields LIKE '%holder_id%')
          |   ) b
          |ON    a.rowkey = b.rowkey
          |)
@@ -583,47 +583,4 @@ object inc_company_relation_v2 {
     re.exportESPre()
     spark.stop()
   }
-}
-
-//    sql(
-//      s"""
-//         |SELECT  a.*
-//         |FROM (
-//         |SELECT  *
-//         |FROM    (
-//         |           SELECT  *
-//         |                ,ROW_NUMBER() OVER (PARTITION BY company_id,human_pid ORDER BY ds desc,update_time desc) num
-//         |           FROM  (
-//         |                     SELECT  *
-//         |                     FROM    $inc_ads_company_human_relation
-//         |                     WHERE   ds > '0'
-//         |                     -- AND     deleted <> 9
-//         |                     UNION ALL
-//         |                     SELECT  *
-//         |                     FROM    $inc_ads_company_human_relation_merge
-//         |                     WHERE   ds > '0'
-//         |                     -- AND     deleted <> 9
-//         |              )
-//         |          )
-//         |WHERE   num = 1
-//         |) a
-//         |JOIN (
-//         |              SELECT  start_id
-//         |              FROM    (
-//         |                          SELECT  start_id
-//         |                          FROM    $inc_ads_relation_holder
-//         |                          WHERE   ds = '$ds'
-//         |                          AND     holder_type = 1
-//         |                          UNION ALL
-//         |                          SELECT  start_id
-//         |                          FROM    $inc_ads_relation_staff
-//         |                          WHERE   ds = '$ds'
-//         |                          UNION ALL
-//         |                          SELECT  start_id
-//         |                          FROM    $inc_ads_relation_legal_entity
-//         |                          WHERE   ds = '$ds'
-//         |                          AND     legal_entity_type = 1
-//         |                      )
-//         |              GROUP BY start_id
-//         | ) b on a.human_pid = b.start_id
-//         |""".stripMargin).createOrReplaceTempView("update_tab")
+}