feat: 信用惩戒查询新需求

src/main/scala/com/winhc/bigdata/spark/implicits/BaseHelper.scala

@@ -29,4 +29,10 @@ object BaseHelper {
     }
   }
 
+
+  implicit class SeqEnhancer(arr: Seq[String]) extends Serializable {
+    def mkStringOrNull(): String = {
+      if (arr.isEmpty) null else arr.mkString(",")
+    }
+  }
 }

src/main/scala/com/winhc/bigdata/spark/ng/credit_punishment/CreditPunishmentCaseAgg.scala

@@ -40,6 +40,8 @@ case class CreditPunishmentCaseAgg(s: SparkSession
 
     spark.udf.register("generate_entity_id", generate_entity_id _)
 
+    def keyno_type(keyno: String): String = if (keyno == null) "1" else if (keyno.length == 32) "0" else "1"
+    spark.udf.register("keyno_type", keyno_type _)
 
     cleanup()
     sql(
@@ -53,19 +55,30 @@ case class CreditPunishmentCaseAgg(s: SparkSession
          |    ,case_no STRING COMMENT '案号'
          |    ,rowkey STRING COMMENT '涉案rowkey'
          |    ,card_num STRING COMMENT '案号'
-         |    ,keyno STRING COMMENT '惩戒主体id'
+         |    ,keyno STRING COMMENT '惩戒主体id，公司id或人id'
+         |    ,keyno_type STRING COMMENT '惩戒主体类型，0 公司 1人'
          |    ,label STRING COMMENT '标签'
          |    ,case_create_time STRING COMMENT '立案时间'
          |    ,record_num bigint COMMENT '记录条数'
+         |
+         |    ,exec_info string COMMENT '执行情况'
+         |
          |    ,total_exec_amount STRING COMMENT '累计被执行总金额'
          |    ,total_no_exec_amount STRING COMMENT '疑似当前欠款总金额'
          |    ,zxr_total_exec_amount STRING COMMENT '被执行人当前被执行总金额'
          |    ,final_case_exec_amount STRING COMMENT '终本案件执行标的总金额'
          |    ,final_case_no_exec_amount STRING COMMENT '终本案件未履行总金额'
-         |    ,company_dishonest_info_num BIGINT COMMENT '失信人条数'
-         |    ,company_zxr_num BIGINT COMMENT '被执条数'
-         |    ,company_zxr_final_case_num BIGINT COMMENT '终本案件条数'
-         |    ,company_zxr_restrict_num BIGINT COMMENT '限高条数'
+         |
+         |    ,company_dishonest_info_num_0 BIGINT COMMENT '失信人条数'
+         |    ,company_zxr_num_0 BIGINT COMMENT '被执条数'
+         |    ,company_zxr_final_case_num_0 BIGINT COMMENT '终本案件条数'
+         |    ,company_zxr_restrict_num_0 BIGINT COMMENT '限高条数'
+         |
+         |    ,company_dishonest_info_num_1 BIGINT COMMENT '失信人条数-历史'
+         |    ,company_zxr_num_1 BIGINT COMMENT '被执条数-历史'
+         |    ,company_zxr_final_case_num_1 BIGINT COMMENT '终本案件条数-历史'
+         |    ,company_zxr_restrict_num_1 BIGINT COMMENT '限高条数-历史'
+         |
          |    ,deleted BIGINT COMMENT '0：未删除，1：删除'
          |    ,all_rowkey STRING COMMENT '所有rowkey的集合，包含历史非历史'
          |)
@@ -129,7 +142,7 @@ case class CreditPunishmentCaseAgg(s: SparkSession
       s"""
          |SELECT
          |        split(xjk_rowkey,'@@') [0 ] AS tn
-         |        ,split(xjk_rowkey,'@@') [1 ] AS rowkey
+         |        ,split(xjk_rowkey,'@@') [2 ] AS rowkey
          |FROM    (
          |        SELECT
          |                *
@@ -196,90 +209,76 @@ case class CreditPunishmentCaseAgg(s: SparkSession
     "all_record"
   }
 
+  private def get_sql(org_tab: String): String = {
+    s"""
+       |SELECT  md5(cleanup(CONCAT_WS('',name,court_name,case_no))) AS credit_punishment_case_id
+       |        ,generate_entity_id(mm['keyno'],mm['card_num'],md5(CONCAT_WS('',mm['card_num'],name)),md5(cleanup(CONCAT_WS('',name,court_name,case_no)))) as credit_punishment_entity_id
+       |        ,name
+       |        ,court_name
+       |        ,case_no
+       |        ,mm['rowkey'] AS rowkey
+       |        ,mm['card_num'] AS card_num
+       |        ,mm['keyno'] AS keyno
+       |        ,keyno_type(mm['keyno']) AS keyno_type
+       |        ,mm['label'] AS label
+       |        ,mm['case_create_time'] AS case_create_time
+       |        ,cast(mm['record_num'] as bigint) AS record_num
+       |
+       |        ,mm['exec_info'] AS exec_info
+       |
+       |        ,mm['total_exec_amount'] AS total_exec_amount
+       |        ,mm['total_no_exec_amount'] AS total_no_exec_amount
+       |        ,mm['zxr_total_exec_amount'] AS zxr_total_exec_amount
+       |        ,mm['final_case_exec_amount'] AS final_case_exec_amount
+       |        ,mm['final_case_no_exec_amount'] AS final_case_no_exec_amount
+       |
+       |        ,CAST(mm['company_dishonest_info_num_0'] AS BIGINT) AS company_dishonest_info_num_0
+       |        ,CAST(mm['company_zxr_num_0'] AS BIGINT) AS company_zxr_num_0
+       |        ,CAST(mm['company_zxr_final_case_num_0'] AS BIGINT) AS company_zxr_final_case_num_0
+       |        ,CAST(mm['company_zxr_restrict_num_0'] AS BIGINT) AS company_zxr_restrict_num_0
+       |
+       |        ,CAST(mm['company_dishonest_info_num_1'] AS BIGINT) AS company_dishonest_info_num_1
+       |        ,CAST(mm['company_zxr_num_1'] AS BIGINT) AS company_zxr_num_1
+       |        ,CAST(mm['company_zxr_final_case_num_1'] AS BIGINT) AS company_zxr_final_case_num_1
+       |        ,CAST(mm['company_zxr_restrict_num_1'] AS BIGINT) AS company_zxr_restrict_num_1
+       |
+       |        ,cast(mm['deleted'] as bigint) AS deleted
+       |        ,mm['all_rowkey'] AS all_rowkey
+       |FROM    (
+       |            SELECT  name
+       |                    ,court_name
+       |                    ,case_no
+       |                    ,credit_punishment_case_agg(rowkey,tn,keyno,card_num,case_create_time,deleted,detail_data) AS mm
+       |            FROM    $org_tab
+       |            GROUP BY name
+       |                     ,court_name
+       |                     ,case_no
+       |        )
+       |WHERE   name IS NOT NULL and trim(name) <> ''
+       |""".stripMargin
+  }
+
 
   def calc(ds: String): Unit = {
     val args = StartAndEndDsUtils(spark).get_gt_ds(CreditPunishment.extractionTargetTab, CreditPunishment.caseTargetTab)
 
     val org_tab = getCreditPunishmentDataExtraction(args.inc_tab_gt_ds)
+
+
     sql(
       s"""
          |---INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE $target_tab PARTITION(ds='${ds}')
-         |SELECT  md5(cleanup(CONCAT_WS('',name,court_name,case_no))) AS credit_punishment_case_id
-         |        ,generate_entity_id(mm['keyno'],mm['card_num'],md5(CONCAT_WS('',mm['card_num'],name)),md5(cleanup(CONCAT_WS('',name,court_name,case_no)))) as credit_punishment_entity_id
-         |        ,name
-         |        ,court_name
-         |        ,case_no
-         |        ,mm['rowkey'] AS rowkey
-         |        ,mm['card_num'] AS card_num
-         |        ,mm['keyno'] AS keyno
-         |        ,mm['label'] AS label
-         |        ,mm['case_create_time'] AS case_create_time
-         |        ,cast(mm['record_num'] as bigint) AS record_num
-         |        ,mm['total_exec_amount'] AS total_exec_amount
-         |        ,mm['total_no_exec_amount'] AS total_no_exec_amount
-         |        ,mm['zxr_total_exec_amount'] AS zxr_total_exec_amount
-         |        ,mm['final_case_exec_amount'] AS final_case_exec_amount
-         |        ,mm['final_case_no_exec_amount'] AS final_case_no_exec_amount
-         |        ,CAST(mm['company_dishonest_info_num'] AS BIGINT) AS company_dishonest_info_num
-         |        ,CAST(mm['company_zxr_num'] AS BIGINT) AS company_zxr_num
-         |        ,CAST(mm['company_zxr_final_case_num'] AS BIGINT) AS company_zxr_final_case_num
-         |        ,CAST(mm['company_zxr_restrict_num'] AS BIGINT) AS company_zxr_restrict_num
-         |        ,cast(mm['deleted'] as bigint) AS deleted
-         |        ,mm['all_rowkey'] AS all_rowkey
-         |FROM    (
-         |            SELECT  name
-         |                    ,court_name
-         |                    ,case_no
-         |                    ,credit_punishment_case_agg(rowkey,tn,keyno,card_num,case_create_time,deleted,detail_data) AS mm
-         |            FROM    $org_tab
-         |            GROUP BY name
-         |                     ,court_name
-         |                     ,case_no
-         |        )
-         |WHERE   name IS NOT NULL and trim(name) <> ''
+         |${get_sql(org_tab)}
          |""".stripMargin)
       .cache()
       .createTempView(s"tmp_data1")
 
     val all_record = getAllDataByTmpData("tmp_data1")
 
-
     sql(
       s"""
          |INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE $target_tab PARTITION(ds='${ds}')
-         |SELECT  md5(cleanup(CONCAT_WS('',name,court_name,case_no))) AS credit_punishment_case_id
-         |        ,generate_entity_id(mm['keyno'],mm['card_num'],md5(CONCAT_WS('',mm['card_num'],name)),md5(cleanup(CONCAT_WS('',name,court_name,case_no)))) as credit_punishment_entity_id
-         |        ,name
-         |        ,court_name
-         |        ,case_no
-         |        ,mm['rowkey'] AS rowkey
-         |        ,mm['card_num'] AS card_num
-         |        ,mm['keyno'] AS keyno
-         |        ,mm['label'] AS label
-         |        ,mm['case_create_time'] AS case_create_time
-         |        ,cast(mm['record_num'] as bigint) AS record_num
-         |        ,mm['total_exec_amount'] AS total_exec_amount
-         |        ,mm['total_no_exec_amount'] AS total_no_exec_amount
-         |        ,mm['zxr_total_exec_amount'] AS zxr_total_exec_amount
-         |        ,mm['final_case_exec_amount'] AS final_case_exec_amount
-         |        ,mm['final_case_no_exec_amount'] AS final_case_no_exec_amount
-         |        ,CAST(mm['company_dishonest_info_num'] AS BIGINT) AS company_dishonest_info_num
-         |        ,CAST(mm['company_zxr_num'] AS BIGINT) AS company_zxr_num
-         |        ,CAST(mm['company_zxr_final_case_num'] AS BIGINT) AS company_zxr_final_case_num
-         |        ,CAST(mm['company_zxr_restrict_num'] AS BIGINT) AS company_zxr_restrict_num
-         |        ,cast(mm['deleted'] as bigint) AS deleted
-         |        ,mm['all_rowkey'] AS all_rowkey
-         |FROM    (
-         |            SELECT  name
-         |                    ,court_name
-         |                    ,case_no
-         |                    ,credit_punishment_case_agg(rowkey,tn,keyno,card_num,case_create_time,deleted,detail_data) AS mm
-         |            FROM    $all_record
-         |            GROUP BY name
-         |                     ,court_name
-         |                     ,case_no
-         |        )
-         |WHERE   name IS NOT NULL and trim(name) <> ''
+         |${get_sql(all_record)}
          |""".stripMargin)
   }
 }

src/main/scala/com/winhc/bigdata/spark/ng/credit_punishment/CreditPunishmentEntityAgg.scala

@@ -66,7 +66,7 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |CREATE TABLE IF NOT EXISTS $target_tab
          |(
          |    credit_punishment_entity_id STRING COMMENT '信用惩戒对象主体id'
-         |    ,keyno STRING COMMENT '惩戒主体id'
+         |    ,keyno STRING COMMENT '惩戒主体id，公司id或人id'
          |    ,name STRING COMMENT '惩戒主体名称'
          |    ,entity_type bigint COMMENT '0为公司，1为人'
          |    ,card_num STRING COMMENT '身份证号或工商注册号'
@@ -81,10 +81,17 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |    ,zxr_total_exec_amount STRING COMMENT '被执行人当前被执行总金额'
          |    ,final_case_exec_amount STRING COMMENT '终本案件执行标的总金额'
          |    ,final_case_no_exec_amount STRING COMMENT '终本案件未履行总金额'
-         |    ,company_dishonest_info_num BIGINT COMMENT '失信人条数'
-         |    ,company_zxr_num BIGINT COMMENT '被执条数'
-         |    ,company_zxr_final_case_num BIGINT COMMENT '终本案件条数'
-         |    ,company_zxr_restrict_num BIGINT COMMENT '限高条数'
+         |
+         |    ,company_dishonest_info_num_0 BIGINT COMMENT '失信人条数'
+         |    ,company_zxr_num_0 BIGINT COMMENT '被执条数'
+         |    ,company_zxr_final_case_num_0 BIGINT COMMENT '终本案件条数'
+         |    ,company_zxr_restrict_num_0 BIGINT COMMENT '限高条数'
+         |
+         |    ,company_dishonest_info_num_1 BIGINT COMMENT '失信人条数-历史'
+         |    ,company_zxr_num_1 BIGINT COMMENT '被执条数-历史'
+         |    ,company_zxr_final_case_num_1 BIGINT COMMENT '终本案件条数-历史'
+         |    ,company_zxr_restrict_num_1 BIGINT COMMENT '限高条数-历史'
+         |
          |    ,deleted BIGINT COMMENT '0：未删除，1：删除'
          |)
          |COMMENT '信用惩戒查询-惩戒主体表'
@@ -184,10 +191,17 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |        ,double_2_str(sum(CAST(zxr_total_exec_amount AS DOUBLE) )) AS zxr_total_exec_amount
          |        ,double_2_str(sum(CAST(final_case_exec_amount AS DOUBLE) )) AS final_case_exec_amount
          |        ,double_2_str(sum(CAST(final_case_no_exec_amount AS DOUBLE) )) AS final_case_no_exec_amount
-         |        ,sum(company_dishonest_info_num) as company_dishonest_info_num
-         |        ,sum(company_zxr_num) as company_zxr_num
-         |        ,sum(company_zxr_final_case_num) as company_zxr_final_case_num
-         |        ,sum(company_zxr_restrict_num) as company_zxr_restrict_num
+         |
+         |        ,sum(company_dishonest_info_num_0) as company_dishonest_info_num_0
+         |        ,sum(company_zxr_num_0) as company_zxr_num_0
+         |        ,sum(company_zxr_final_case_num_0) as company_zxr_final_case_num_0
+         |        ,sum(company_zxr_restrict_num_0) as company_zxr_restrict_num_0
+         |
+         |        ,sum(company_dishonest_info_num_1) as company_dishonest_info_num_1
+         |        ,sum(company_zxr_num_1) as company_zxr_num_1
+         |        ,sum(company_zxr_final_case_num_1) as company_zxr_final_case_num_1
+         |        ,sum(company_zxr_restrict_num_1) as company_zxr_restrict_num_1
+         |
          |        ,deleted_merge(deleted) AS deleted
          |FROM    $org_tab
          |GROUP BY credit_punishment_entity_id
@@ -215,10 +229,17 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |        ,t1.zxr_total_exec_amount
          |        ,t1.final_case_exec_amount
          |        ,t1.final_case_no_exec_amount
-         |        ,t1.company_dishonest_info_num
-         |        ,t1.company_zxr_num
-         |        ,t1.company_zxr_final_case_num
-         |        ,t1.company_zxr_restrict_num
+         |
+         |        ,t1.company_dishonest_info_num_0
+         |        ,t1.company_zxr_num_0
+         |        ,t1.company_zxr_final_case_num_0
+         |        ,t1.company_zxr_restrict_num_0
+         |
+         |        ,t1.company_dishonest_info_num_1
+         |        ,t1.company_zxr_num_1
+         |        ,t1.company_zxr_final_case_num_1
+         |        ,t1.company_zxr_restrict_num_1
+         |
          |        ,t1.deleted
          |FROM    (
          |            SELECT  *
@@ -245,10 +266,17 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |        ,zxr_total_exec_amount
          |        ,final_case_exec_amount
          |        ,final_case_no_exec_amount
-         |        ,company_dishonest_info_num
-         |        ,company_zxr_num
-         |        ,company_zxr_final_case_num
-         |        ,company_zxr_restrict_num
+         |
+         |        ,company_dishonest_info_num_0
+         |        ,company_zxr_num_0
+         |        ,company_zxr_final_case_num_0
+         |        ,company_zxr_restrict_num_0
+         |
+         |        ,company_dishonest_info_num_1
+         |        ,company_zxr_num_1
+         |        ,company_zxr_final_case_num_1
+         |        ,company_zxr_restrict_num_1
+         |
          |        ,deleted
          |FROM    all_data
          |WHERE   not (keyno is not null and length(keyno) = 32) and card_num is not null
@@ -271,10 +299,17 @@ case class CreditPunishmentEntityAgg(s: SparkSession
          |        ,zxr_total_exec_amount
          |        ,final_case_exec_amount
          |        ,final_case_no_exec_amount
-         |        ,company_dishonest_info_num
-         |        ,company_zxr_num
-         |        ,company_zxr_final_case_num
-         |        ,company_zxr_restrict_num
+         |
+         |        ,company_dishonest_info_num_0
+         |        ,company_zxr_num_0
+         |        ,company_zxr_final_case_num_0
+         |        ,company_zxr_restrict_num_0
+         |
+         |        ,company_dishonest_info_num_1
+         |        ,company_zxr_num_1
+         |        ,company_zxr_final_case_num_1
+         |        ,company_zxr_restrict_num_1
+         |
          |        ,deleted
          |FROM    all_data
          |WHERE   not (keyno is not null and length(keyno) = 32) and card_num is null

src/main/scala/com/winhc/bigdata/spark/ng/credit_punishment/udf/CreditPunishmentCaseAggUDF.scala

@@ -1,6 +1,8 @@
 package com.winhc.bigdata.spark.ng.credit_punishment.udf
 
+import cn.hutool.core.util.NumberUtil
 import com.alibaba.fastjson.{JSON, JSONPath}
+import com.winhc.bigdata.spark.implicits.BaseHelper._
 import com.winhc.bigdata.spark.utils.{BaseUtil, RegCapitalAmount}
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.sql.Row
@@ -56,7 +58,10 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
     , StructField("zxr_total_exec_amount", DoubleType) //被执行人当前被执行总金额 7
     , StructField("final_case_exec_amount", DoubleType) //终本案件执行标的总金额 8
     , StructField("final_case_no_exec_amount", DoubleType) //终本案件未履行总金额 9
-    , StructField("all_rowkey", ArrayType(StringType, containsNull = false)) //所有rowkey的集合，包含历史非历史
+    , StructField("all_rowkey", ArrayType(StringType, containsNull = false)) //所有rowkey的集合，包含历史非历史 10
+
+    , StructField("final_case_exec_info", StringType) //11
+    , StructField("dishonest_exec_info", StringType) //12
   ))
 
   override def dataType: DataType = DataTypes.createMapType(StringType, StringType)
@@ -75,6 +80,9 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
     buffer.update(8, 0d)
     buffer.update(9, 0d)
     buffer.update(10, Seq.empty[String])
+
+    buffer.update(11, null)
+    buffer.update(12, null)
   }
 
   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
@@ -86,7 +94,7 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
     val deleted = input.getLong(5)
     val detail_data = input.getString(6)
 
-    buffer(10) = s"$tn$delimiter$rowkey" +: buffer.getSeq[String](0)
+    buffer(10) = s"$tn$delimiter$deleted$delimiter$rowkey" +: buffer.getSeq[String](0)
 
     deleted match {
       //不计历史信息：
@@ -97,6 +105,23 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
         buffer(3) = tn +: buffer.getSeq[String](3)
         buffer(4) = getCaseCreateTime(buffer.getTimestamp(4), case_create_time)
         buffer(6) = buffer.getDouble(6) + getAmount(detail_data, "$.no_exec_amount")
+
+        if (tn.equals("company_zxr_final_case")) {
+          val exec_money = getAmount(detail_data, "$.exec_amount")
+          val no_exec_amount = getAmount(detail_data, "$.no_exec_amount")
+          if (exec_money != 0 && no_exec_amount != 0) {
+            var str: String = null
+            if (no_exec_amount > exec_money) {
+              str = "履行0%"
+            } else {
+              str = NumberUtil.decimalFormat("履行#.##%", NumberUtil.sub(1, NumberUtil.div(no_exec_amount, exec_money)))
+            }
+            buffer(11) = str
+          }
+        }
+        if (tn.equals("company_dishonest_info")) {
+          buffer(12) = getStr(detail_data, "$.performance")
+        }
       }
       case 1 => {
       }
@@ -109,7 +134,7 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
         buffer(7) = buffer.getDouble(7) + getAmount(detail_data, "$.exec_money")
       }
       case "company_zxr_final_case" => {
-        buffer(8) = buffer.getDouble(8) + getAmount(detail_data, "$.exec_money")
+        buffer(8) = buffer.getDouble(8) + getAmount(detail_data, "$.exec_amount")
         buffer(9) = buffer.getDouble(9) + getAmount(detail_data, "$.no_exec_amount")
       }
       case _ => {}
@@ -129,12 +154,23 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
     buffer1(8) = buffer1.getDouble(8) + buffer2.getDouble(8)
     buffer1(9) = buffer1.getDouble(9) + buffer2.getDouble(9)
     buffer1(10) = buffer1.getSeq[String](10) ++ buffer2.getSeq[String](10)
+
+    if (buffer1.getString(11) == null) {
+      buffer1(11) = buffer2.getString(11)
+    }
+    if (buffer1.getString(12) == null) {
+      buffer1(12) = buffer2.getString(12)
+    }
   }
 
   override def evaluate(buffer: Row): Any = {
     val strings = buffer.getSeq[String](0).distinct.filter(StringUtils.isNotBlank)
+    val all_rowkey = buffer.getSeq[String](10).distinct.filter(StringUtils.isNotBlank)
 
-    val tnDistribution: Map[String, Int] = strings.map(r => r.split(delimiter)(0)).groupBy(f => f).mapValues(_.size)
+    val stringToInt: Map[String, Int] = all_rowkey.map(r => {
+      val ss = r.split(delimiter)
+      (ss(0), ss(1))
+    }).filter(r => r._2.equals("0") || r._2.equals("1")).map(r => s"${r._1}_${r._2}").groupBy(f => f).mapValues(_.size)
 
     val rowkey = strings
       .map(r => {
@@ -156,33 +192,49 @@ case class CreditPunishmentCaseAggUDF() extends UserDefinedAggregateFunction {
 
     val cct = if (case_create_time == null) null else df.format(case_create_time.toInstant)
     val deleted = if (rowkey.isEmpty) "1" else "0"
+
+    val e1 = buffer.getString(11)
+    val e2 = buffer.getString(12)
+
+    val exec_info = if (e1 == null) if (e2 == null) "未知" else e2 else e1
     Map(
       "rowkey" -> rowkey.mkString(",")
       , "card_num" -> card_num.headOption.getOrElse(null)
       , "keyno" -> keyno.headOption.getOrElse(null)
-      , "label" -> label.mkString(",")
+      , "label" -> label.mkStringOrNull()
       , "case_create_time" -> cct
       , "record_num" -> s"${rowkey.length}"
+
+      , "exec_info" -> exec_info
+
       , "total_exec_amount" -> double2String(total_exec_amount)
       , "total_no_exec_amount" -> double2String(total_no_exec_amount)
       , "zxr_total_exec_amount" -> double2String(zxr_total_exec_amount)
       , "final_case_exec_amount" -> double2String(final_case_exec_amount)
       , "final_case_no_exec_amount" -> double2String(final_case_no_exec_amount)
 
-      , "company_dishonest_info_num" -> s"${tnDistribution.getOrElse("company_dishonest_info", 0)}"
-      , "company_zxr_num" -> s"${tnDistribution.getOrElse("company_zxr", 0)}"
-      , "company_zxr_final_case_num" -> s"${tnDistribution.getOrElse("company_zxr_final_case", 0)}"
-      , "company_zxr_restrict_num" -> s"${tnDistribution.getOrElse("company_zxr_restrict", 0)}"
+      , "company_dishonest_info_num_0" -> s"${stringToInt.getOrElse("company_dishonest_info_0", 0)}"
+      , "company_zxr_num_0" -> s"${stringToInt.getOrElse("company_zxr_0", 0)}"
+      , "company_zxr_final_case_num_0" -> s"${stringToInt.getOrElse("company_zxr_final_case_0", 0)}"
+      , "company_zxr_restrict_num_0" -> s"${stringToInt.getOrElse("company_zxr_restrict_0", 0)}"
+
+      , "company_dishonest_info_num_1" -> s"${stringToInt.getOrElse("company_dishonest_info_1", 0)}"
+      , "company_zxr_num_1" -> s"${stringToInt.getOrElse("company_zxr_1", 0)}"
+      , "company_zxr_final_case_num_1" -> s"${stringToInt.getOrElse("company_zxr_final_case_1", 0)}"
+      , "company_zxr_restrict_num_1" -> s"${stringToInt.getOrElse("company_zxr_restrict_1", 0)}"
 
       , "deleted" -> deleted
-      , "all_rowkey" -> buffer.getSeq[String](10).distinct.filter(StringUtils.isNotBlank).mkString(",")
+      , "all_rowkey" -> all_rowkey.mkString(",")
     )
 
   }
 
+  private def getStr(json: String, jsonPath: String): String = {
+    JSONPath.eval(JSON.parseObject(json), jsonPath).asInstanceOf[String]
+  }
 
   private def getAmount(json: String, jsonPath: String): Double = {
-    val str = JSONPath.eval(JSON.parseObject(json), jsonPath).asInstanceOf[String]
+    val str = getStr(json, jsonPath)
     try {
       RegCapitalAmount.getAmount(str).toDouble / 100
     } catch {