|
|
@@ -0,0 +1,304 @@
|
|
|
+package com.winhc.bigdata.spark.ng.utils
|
|
|
+
|
|
|
+import com.winhc.bigdata.spark.utils.{BaseUtil, LoggingUtils, SparkUtils}
|
|
|
+import org.apache.commons.lang3.StringUtils
|
|
|
+import org.apache.spark.sql.SparkSession
|
|
|
+import org.apache.spark.sql.functions.{col, struct, to_json}
|
|
|
+
|
|
|
+import scala.annotation.meta.getter
|
|
|
+import scala.collection.mutable
|
|
|
+
|
|
|
+/**
|
|
|
+ * @Author: XuJiakai
|
|
|
+ * @Date: 2020/10/10 10:23
|
|
|
+ * @Description: winhc_ng空间下摘要增强版
|
|
|
+ */
|
|
|
+case class GroupByInfoNg(field: String, value_alias: Seq[(String, String)])
|
|
|
+
|
|
|
+case class company_summary_args(table_name: String
|
|
|
+ , companyIdField: String
|
|
|
+ , field_prefix: String = null //如果没有 groupByInfo的话可使用此别名，默认为table_name
|
|
|
+ , distinctField: String = "rowkey"
|
|
|
+ , groupByInfo: GroupByInfoNg = null
|
|
|
+ , where: String = ""
|
|
|
+ , sortField: String = "ds"
|
|
|
+ ){
|
|
|
+ def winhc_hash(): String ={
|
|
|
+ BaseUtil.cleanup(s"${table_name}_${companyIdField}").replaceAll("[\\(\\)（） ]","")
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+case class CompanySummaryNg_new(s: SparkSession,
|
|
|
+ project: String //表所在工程名
|
|
|
+ , args: Seq[company_summary_args]
|
|
|
+ ) extends LoggingUtils {
|
|
|
+ @(transient@getter) val spark: SparkSession = s
|
|
|
+ private val target_tab = "winhc_ng.out_es_summary_test"
|
|
|
+ init()
|
|
|
+
|
|
|
+ private def init() {
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |CREATE TABLE IF NOT EXISTS $target_tab
|
|
|
+ |(
|
|
|
+ | company_id STRING COMMENT '公司id'
|
|
|
+ | ,summary STRING COMMENT '格子中的摘要信息，json格式'
|
|
|
+ | ,detail STRING COMMENT '个别维度详细的摘要信息'
|
|
|
+ |)
|
|
|
+ |COMMENT 'out es summary,create by ${BaseUtil.nowDate(pattern = "yyyy-MM-dd HH:mm:ss")}'
|
|
|
+ |PARTITIONED BY (ds STRING COMMENT '分区')
|
|
|
+ |LIFECYCLE 15
|
|
|
+ |""".stripMargin)
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private def get_table_data(arg: company_summary_args): String = {
|
|
|
+ val ds = getLastPartitionsOrElse(target_tab, null)
|
|
|
+
|
|
|
+ val tab = arg.table_name
|
|
|
+ val companyIdField = arg.companyIdField
|
|
|
+ val distinctField = arg.distinctField
|
|
|
+
|
|
|
+ val sortField = arg.sortField
|
|
|
+ val where = if (StringUtils.isEmpty(arg.where)) {
|
|
|
+ s""
|
|
|
+ } else {
|
|
|
+ s"AND ${arg.where}"
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ val ads_table = s"${project}.ads_$tab" //存量ads表
|
|
|
+ val inc_ads_table = s"${project}.inc_ads_$tab"
|
|
|
+
|
|
|
+ val new_cols = getColumns(ads_table).intersect(getColumns(inc_ads_table))
|
|
|
+
|
|
|
+ val ads_last_ds = getLastPartitionsOrElse(ads_table, "0")
|
|
|
+ val tab_tmp_view = s"${arg.winhc_hash()}_data"
|
|
|
+
|
|
|
+ if (ds == null) {
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ |FROM $ads_table
|
|
|
+ |WHERE ds = '$ads_last_ds'
|
|
|
+ |UNION ALL
|
|
|
+ |SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ |FROM $inc_ads_table
|
|
|
+ |WHERE ds > $ads_last_ds
|
|
|
+ |""".stripMargin)
|
|
|
+ .createOrReplaceTempView(tab_tmp_view)
|
|
|
+ } else {
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |SELECT ${new_cols.map(getCastCols(_, "org_tab.")).mkString(",")}
|
|
|
+ |FROM (
|
|
|
+ | SELECT DISTINCT $companyIdField as xjk_cid
|
|
|
+ | FROM $inc_ads_table
|
|
|
+ | WHERE ds = $ds
|
|
|
+ | ) id_table
|
|
|
+ |JOIN (
|
|
|
+ | SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ | ,$companyIdField as xjk_cid
|
|
|
+ | FROM $inc_ads_table
|
|
|
+ | WHERE ds > '$ads_last_ds'
|
|
|
+ | AND ds < '$ds'
|
|
|
+ | UNION ALL
|
|
|
+ | SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ | ,$companyIdField as xjk_cid
|
|
|
+ | FROM $ads_table
|
|
|
+ | WHERE ds = '$ads_last_ds'
|
|
|
+ | ) org_tab
|
|
|
+ |ON id_table.xjk_cid = org_tab.xjk_cid
|
|
|
+ |UNION ALL
|
|
|
+ |SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ |FROM $inc_ads_table
|
|
|
+ |WHERE ds = $ds
|
|
|
+ |""".stripMargin)
|
|
|
+ .createOrReplaceTempView(tab_tmp_view)
|
|
|
+ }
|
|
|
+
|
|
|
+ val distinct_tab = s"${tab_tmp_view}_distinct"
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |SELECT ${new_cols.map(getCastCols(_, "")).mkString(",")}
|
|
|
+ |FROM (
|
|
|
+ | SELECT tmp.*
|
|
|
+ | ,ROW_NUMBER() OVER(PARTITION BY $distinctField ORDER BY $sortField DESC ) c
|
|
|
+ | FROM $tab_tmp_view AS tmp
|
|
|
+ | ) tmp2
|
|
|
+ |WHERE tmp2.c = 1
|
|
|
+ |$where
|
|
|
+ |""".stripMargin)
|
|
|
+ .createOrReplaceTempView(distinct_tab)
|
|
|
+ distinct_tab
|
|
|
+ }
|
|
|
+
|
|
|
+ private def get_tab_summary(arg: company_summary_args): String = {
|
|
|
+ val tab_hash = arg.winhc_hash()
|
|
|
+
|
|
|
+ val tab = arg.table_name
|
|
|
+ val companyIdField = arg.companyIdField
|
|
|
+
|
|
|
+ val result_tab = s"${tab_hash}_summary_tab"
|
|
|
+
|
|
|
+ val all_tab = get_table_data(arg)
|
|
|
+
|
|
|
+ val func_name = s"xjk_func_${tab_hash}"
|
|
|
+
|
|
|
+ val view = if (arg.groupByInfo == null) {
|
|
|
+ if (arg.field_prefix != null) {
|
|
|
+ s"arr[0] as ${arg.field_prefix}"
|
|
|
+ } else {
|
|
|
+ s"arr[0] as $tab"
|
|
|
+ }
|
|
|
+
|
|
|
+ } else {
|
|
|
+ arg.groupByInfo.value_alias.indices.map(i => {
|
|
|
+ s"arr[$i] as ${arg.groupByInfo.value_alias(i)._2}"
|
|
|
+ }).mkString(",")
|
|
|
+ }
|
|
|
+
|
|
|
+ //注册函数
|
|
|
+ if (arg.groupByInfo != null) {
|
|
|
+ val fieldSeq = arg.groupByInfo.value_alias.map(r => {
|
|
|
+ (s"${r._1}", r._2)
|
|
|
+ })
|
|
|
+
|
|
|
+ def getResArr(group_val: String, num: Long): Seq[Long] = {
|
|
|
+ val res = scala.collection.mutable.ArrayBuffer[Long]()
|
|
|
+ for (i <- fieldSeq) {
|
|
|
+ if (i._1.equals(group_val)) {
|
|
|
+ res += num
|
|
|
+ } else {
|
|
|
+ res += 0
|
|
|
+ }
|
|
|
+ }
|
|
|
+ res
|
|
|
+ }
|
|
|
+
|
|
|
+ spark.udf.register(func_name, getResArr _)
|
|
|
+ }
|
|
|
+
|
|
|
+ val groupKey_show = if (arg.groupByInfo == null) {
|
|
|
+ s",array(count(1)) as arr"
|
|
|
+ } else {
|
|
|
+ s",$func_name(cast(${arg.groupByInfo.field} as STRING),count(1)) as arr"
|
|
|
+ }
|
|
|
+
|
|
|
+ val groupKey = if (arg.groupByInfo == null) {
|
|
|
+ s""
|
|
|
+ } else {
|
|
|
+ s",${arg.groupByInfo.field}"
|
|
|
+ }
|
|
|
+
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |SELECT company_id
|
|
|
+ | ,${view}
|
|
|
+ |FROM (
|
|
|
+ | SELECT $companyIdField as company_id
|
|
|
+ | $groupKey_show
|
|
|
+ | FROM $all_tab
|
|
|
+ | GROUP BY $companyIdField ${groupKey}
|
|
|
+ |)
|
|
|
+ |""".stripMargin)
|
|
|
+ .createOrReplaceTempView(result_tab)
|
|
|
+
|
|
|
+ if (arg.groupByInfo != null) {
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |SELECT company_id
|
|
|
+ | ,${arg.groupByInfo.value_alias.map(_._2).map(f => s"sum($f) as $f").mkString(",")}
|
|
|
+ |FROM $result_tab
|
|
|
+ |GROUP BY company_id
|
|
|
+ |""".stripMargin)
|
|
|
+ .createOrReplaceTempView(result_tab)
|
|
|
+ }
|
|
|
+ result_tab
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ def calc(): Unit = {
|
|
|
+ val summary_tab = "summary_tab_xjk"
|
|
|
+ val summary_tabs = args.map(get_tab_summary).seq
|
|
|
+ merge_table(spark, summary_tabs, "company_id").calc(summary_tab)
|
|
|
+
|
|
|
+ val cols = getColumns(summary_tab).diff(Seq("company_id"))
|
|
|
+
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |select * from $summary_tab
|
|
|
+ |""".stripMargin)
|
|
|
+ .withColumn("summary", to_json(struct(cols.map(col): _*))).createTempView("xjk_tmp_summary_tab")
|
|
|
+
|
|
|
+ sql(
|
|
|
+ s"""
|
|
|
+ |INSERT OVERWRITE TABLE $target_tab PARTITION(ds='${BaseUtil.getYesterday()}')
|
|
|
+ |SELECT company_id,summary,null as detail
|
|
|
+ |FROM
|
|
|
+ | xjk_tmp_summary_tab
|
|
|
+ |""".stripMargin)
|
|
|
+ }
|
|
|
+
|
|
|
+ private def getCastCols(name: String, pre: String): String = {
|
|
|
+ val list = List("cid", "new_cid", "ncid")
|
|
|
+ if (list.contains(name)) {
|
|
|
+ return s"CAST(${pre}${name} as BIGINT) $name"
|
|
|
+ }
|
|
|
+ pre + name
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+object CompanySummaryNg_new {
|
|
|
+
|
|
|
+ private def get_default_summary_args(tableName: String): company_summary_args = {
|
|
|
+ company_summary_args(
|
|
|
+ table_name = tableName
|
|
|
+ , companyIdField = "split(rowkey,'_')[0]"
|
|
|
+ , where = "instr(rowkey,'_') != 0"
|
|
|
+ , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
|
|
|
+ ("0", s"${tableName}_del_0")
|
|
|
+ , ("1", s"${tableName}_del_1")
|
|
|
+ ))
|
|
|
+ )
|
|
|
+ }
|
|
|
+
|
|
|
+ private val start_args = Seq(
|
|
|
+ get_default_summary_args("company_holder")
|
|
|
+ , get_default_summary_args("company_staff")
|
|
|
+ , get_default_summary_args("company_icp")
|
|
|
+ , get_default_summary_args("company_tm")
|
|
|
+ , get_default_summary_args("company_app_info")
|
|
|
+
|
|
|
+ , company_summary_args(table_name = "company_court_open_announcement_explode"
|
|
|
+ , companyIdField = "plaintiff_info_id_explode"
|
|
|
+ , where = "plaintiff_info_id_explode is not null and length(plaintiff_info_id_explode) = 32 and deleted <> 9"
|
|
|
+ , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
|
|
|
+ ("0", s"company_court_open_announcement_deleted_0_plaintiff")
|
|
|
+ , ("1", s"company_court_open_announcement_deleted_1_plaintiff")
|
|
|
+ ))
|
|
|
+ )
|
|
|
+ , company_summary_args(table_name = "company_court_open_announcement_explode"
|
|
|
+ , companyIdField = "defendant_info_id_explode"
|
|
|
+ , where = "defendant_info_id_explode is not null and length(defendant_info_id_explode) = 32 and deleted <> 9"
|
|
|
+ , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
|
|
|
+ ("0", s"company_court_open_announcement_deleted_0_defendant")
|
|
|
+ , ("1", s"company_court_open_announcement_deleted_1_defendant")
|
|
|
+ ))
|
|
|
+ )
|
|
|
+
|
|
|
+ )
|
|
|
+
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+
|
|
|
+ val config = mutable.Map(
|
|
|
+ "spark.hadoop.odps.project.name" -> "winhc_ng",
|
|
|
+ "spark.debug.maxToStringFields" -> "200",
|
|
|
+ "spark.hadoop.odps.spark.local.partition.amt" -> "10000"
|
|
|
+ )
|
|
|
+
|
|
|
+ val spark = SparkUtils.InitEnv(getClass.getSimpleName, config)
|
|
|
+ CompanySummaryNg_new(s = spark, project = "winhc_ng", args = start_args).calc()
|
|
|
+ spark.stop()
|
|
|
+ }
|
|
|
+}
|
许家凯