|
|
@@ -0,0 +1,226 @@
|
|
|
+package com.winhc.bigdata.spark.jobs.judicial
|
|
|
+
|
|
|
+import com.winhc.bigdata.spark.udf.{BaseFunc, CompanyMapping}
|
|
|
+import com.winhc.bigdata.spark.utils.BaseUtil.isWindows
|
|
|
+import com.winhc.bigdata.spark.utils.{BaseUtil, LoggingUtils, SparkUtils}
|
|
|
+import org.apache.spark.sql.SparkSession
|
|
|
+
|
|
|
+import scala.collection.mutable
|
|
|
+
|
|
|
+
|
|
|
+object JudicialCaseRelationPre12 {
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+ val project = "winhc_eci_dev"
|
|
|
+ println(
|
|
|
+ s"""
|
|
|
+ |project: $project
|
|
|
+ |""".stripMargin)
|
|
|
+
|
|
|
+ val config = mutable.Map(
|
|
|
+ "spark.hadoop.odps.project.name" -> s"$project",
|
|
|
+ "spark.hadoop.odps.spark.local.partition.amt" -> "1000"
|
|
|
+ )
|
|
|
+ val spark: SparkSession = SparkUtils.InitEnv(this.getClass.getSimpleName, config)
|
|
|
+ JudicialCaseRelationPre12(spark, project).precalc()
|
|
|
+ spark.stop()
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+//司法案件，终本数据预处理
|
|
|
+case class JudicialCaseRelationPre12(s: SparkSession, project: String
|
|
|
+ ) extends LoggingUtils with CompanyMapping with BaseFunc {
|
|
|
+ override protected val spark: SparkSession = s
|
|
|
+
|
|
|
+ def precalc(): Unit = {
|
|
|
+ prepareFunctions(spark)
|
|
|
+ case_no_trim_udf()
|
|
|
+ //终本案件-企业
|
|
|
+ var lastDsIncAds = BaseUtil.getPartion(s"$project.inc_ads_company_zxr_final_case", spark)
|
|
|
+ spark.sparkContext.setJobDescription(s"处理company_zxr_final_case($lastDsIncAds)")
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |insert ${if (isWindows) "INTO" else "OVERWRITE"} table $project.ads_judicial_case_relation_pre partition(ds='$lastDsIncAds',tn='zxr_final_case')
|
|
|
+ |select
|
|
|
+ | judicase_id
|
|
|
+ | ,flag
|
|
|
+ | ,title
|
|
|
+ | ,case_type
|
|
|
+ | ,case_reason
|
|
|
+ | ,case_no
|
|
|
+ | ,court_name
|
|
|
+ | ,case_stage
|
|
|
+ | ,yg_name
|
|
|
+ | ,bg_name
|
|
|
+ | ,date
|
|
|
+ | ,detail_id
|
|
|
+ | ,case_amt
|
|
|
+ |from (
|
|
|
+ | select
|
|
|
+ | md5(cleanup(case_no)) as judicase_id
|
|
|
+ | ,"6" as flag
|
|
|
+ | ,concat_ws('',name,'终本案件') as title
|
|
|
+ | ,concat_ws('',case_type(case_no)) as case_type
|
|
|
+ | ,NULL as case_reason
|
|
|
+ | ,case_no
|
|
|
+ | ,court_name
|
|
|
+ | ,concat_ws('',case_stage(case_no)) as case_stage
|
|
|
+ | ,NULL as yg_name
|
|
|
+ | ,name as bg_name
|
|
|
+ | ,case_create_time as date
|
|
|
+ | ,rowkey as detail_id
|
|
|
+ | ,exec_amount as case_amt
|
|
|
+ | ,row_number() over(partition by rowkey order by update_time desc) num
|
|
|
+ | from $project.inc_ads_company_zxr_final_case
|
|
|
+ | where length(case_no) > 0 and ds > '0'
|
|
|
+ | )
|
|
|
+ |where num = 1
|
|
|
+ |""".stripMargin).show(10, false)
|
|
|
+
|
|
|
+
|
|
|
+ //终本案件预处理(个人）
|
|
|
+ val columns: Seq[String] = spark.table(s"$project.inc_ads_company_zxr_final_case_person").schema.map(_.name).filter(_!="flag")
|
|
|
+ lastDsIncAds = BaseUtil.getPartion(s"$project.inc_ads_company_zxr_final_case_person", spark)
|
|
|
+ spark.sparkContext.setJobDescription(s"处理zxr_final_case_person($lastDsIncAds)")
|
|
|
+ //1、先从被执行人中用name和case_no关联补全身份证号码
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |--先从company_zxr_final_case_person存量、增量表获取
|
|
|
+ |SELECT rowkey
|
|
|
+ | ,IF(A.identity_num IS NULL AND B.card IS NOT NULL,1,0) AS flag
|
|
|
+ | ,new_cid
|
|
|
+ | ,cid
|
|
|
+ | ,id
|
|
|
+ | ,name
|
|
|
+ | ,sex
|
|
|
+ | ,COALESCE(A.identity_num,B.card) AS identity_num
|
|
|
+ | ,court_name
|
|
|
+ | ,case_final_time
|
|
|
+ | ,case_create_time
|
|
|
+ | ,A.case_no AS case_no
|
|
|
+ | ,description
|
|
|
+ | ,no_exec_amount
|
|
|
+ | ,exec_amount
|
|
|
+ | ,source
|
|
|
+ | ,status
|
|
|
+ | ,appro_time
|
|
|
+ | ,A.create_time AS create_time
|
|
|
+ | ,A.update_time AS update_time
|
|
|
+ | ,A.deleted AS deleted
|
|
|
+ |FROM(
|
|
|
+ | SELECT *
|
|
|
+ | FROM(
|
|
|
+ | SELECT *
|
|
|
+ | FROM(
|
|
|
+ | SELECT *
|
|
|
+ | ,ROW_NUMBER() OVER (PARTITION BY cleanup(CONCAT_WS('',name,case_no)) ORDER BY update_time DESC ) num
|
|
|
+ | FROM(
|
|
|
+ | SELECT ${columns.mkString(",")}
|
|
|
+ | FROM $project.ads_company_zxr_final_case_person
|
|
|
+ | WHERE ds>'0'
|
|
|
+ | UNION ALL
|
|
|
+ | SELECT ${columns.mkString(",")}
|
|
|
+ | FROM $project.inc_ads_company_zxr_final_case_person
|
|
|
+ | WHERE ds>'0'
|
|
|
+ | )
|
|
|
+ | )
|
|
|
+ | WHERE num=1
|
|
|
+ | )
|
|
|
+ | ) A
|
|
|
+ |LEFT JOIN
|
|
|
+ |(
|
|
|
+ | SELECT *
|
|
|
+ | FROM(
|
|
|
+ | SELECT *
|
|
|
+ | ,ROW_NUMBER() OVER (PARTITION BY cleanup(CONCAT_WS('',cname,case_no)) ORDER BY update_time DESC ) num
|
|
|
+ | FROM (
|
|
|
+ | SELECT cname,card,case_no,type,update_time
|
|
|
+ | FROM $project.ods_company_zxr
|
|
|
+ | WHERE ds>'0'
|
|
|
+ | UNION ALL
|
|
|
+ | SELECT cname,card,case_no,type,update_time
|
|
|
+ | FROM $project.inc_ods_company_zxr
|
|
|
+ | WHERE ds>'0'
|
|
|
+ | )
|
|
|
+ | )
|
|
|
+ | WHERE num=1 AND type='1'
|
|
|
+ |) B
|
|
|
+ |ON A.name=B.cname AND A.case_no=B.case_no
|
|
|
+ |""".stripMargin
|
|
|
+ ).createOrReplaceTempView("tmp_person_cloze_1")
|
|
|
+ //2、再从失信人中用name和case_no关联补全身份证号码
|
|
|
+
|
|
|
+ //3、再从自身表中根据name和company_name去补全身份证号码
|
|
|
+// sql(
|
|
|
+// s"""
|
|
|
+// |--先从前面结果表获取
|
|
|
+// |INSERT OVERWRITE TABLE winhc_eci_dev.ads_company_zxr_final_case_person_cloze PARTITION(ds=$lastDsIncAds)
|
|
|
+// |SELECT A.rowkey
|
|
|
+// | ,IF(A.identity_num IS NULL AND D.identity_num IS NOT NULL,3,0) AS flag
|
|
|
+// | ,A.new_cid
|
|
|
+// | ,A.cid
|
|
|
+// | ,A.id
|
|
|
+// | ,A.name
|
|
|
+// | ,A.sex
|
|
|
+// | ,COALESCE(A.identity_num,D.identity_num) AS identity_num
|
|
|
+// | ,A.court_name
|
|
|
+// | ,A.case_final_time
|
|
|
+// | ,A.case_create_time
|
|
|
+// | ,A.case_no
|
|
|
+// | ,A.description
|
|
|
+// | ,A.no_exec_amount
|
|
|
+// | ,A.exec_amount
|
|
|
+// | ,A.source
|
|
|
+// | ,A.status
|
|
|
+// | ,A.appro_time
|
|
|
+// | ,A.create_time
|
|
|
+// | ,A.update_time
|
|
|
+// | ,A.deleted
|
|
|
+// |FROM tmp_person_cloze_1 A
|
|
|
+// |LEFT JOIN
|
|
|
+// |tmp_person_cloze_1 D
|
|
|
+// |ON A.name=D.name --根据实际数据情况
|
|
|
+// |""".stripMargin
|
|
|
+// )//.createOrReplaceTempView("tmp_person_cloze_3")
|
|
|
+
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |insert ${if (isWindows) "INTO" else "OVERWRITE"} table $project.ads_judicial_case_relation_pre partition(ds='$lastDsIncAds',tn='zxr_final_case_person')
|
|
|
+ |select
|
|
|
+ | judicase_id
|
|
|
+ | ,flag
|
|
|
+ | ,title
|
|
|
+ | ,case_type
|
|
|
+ | ,case_reason
|
|
|
+ | ,case_no
|
|
|
+ | ,court_name
|
|
|
+ | ,case_stage
|
|
|
+ | ,yg_name
|
|
|
+ | ,bg_name
|
|
|
+ | ,date
|
|
|
+ | ,detail_id
|
|
|
+ | ,case_amt
|
|
|
+ |from (
|
|
|
+ | select
|
|
|
+ | md5(cleanup(case_no)) as judicase_id
|
|
|
+ | ,"12" as flag
|
|
|
+ | ,concat_ws('',name,'终本案件') AS title
|
|
|
+ | ,concat_ws('',case_type(case_no)) as case_type
|
|
|
+ | ,NULL AS case_reason
|
|
|
+ | ,case_no
|
|
|
+ | ,court_name
|
|
|
+ | ,concat_ws('',case_stage(case_no)) as case_stage
|
|
|
+ | ,NULL as yg_name
|
|
|
+ | ,name as bg_name
|
|
|
+ | ,case_create_time as date--目前天眼查数据没有执行日期，先以此代替
|
|
|
+ | ,rowkey as detail_id
|
|
|
+ | ,exec_amount as case_amt
|
|
|
+ | ,row_number() over(partition by rowkey order by update_time desc) num
|
|
|
+ | from tmp_person_cloze_1
|
|
|
+ | where length(case_no) > 0
|
|
|
+ | )
|
|
|
+ |where num = 1
|
|
|
+ |""".stripMargin).show(10, false)
|
|
|
+
|
|
|
+ }
|
|
|
+}
|
