|
@@ -117,8 +117,15 @@ case class CompanyIndexJob(s: SparkSession,
|
|
}
|
|
}
|
|
|
|
|
|
private def inc(): Unit = {
|
|
private def inc(): Unit = {
|
|
- val target_ds = getLastPartitionsOrElse(target_tab, null)
|
|
|
|
|
|
+ var target_ds = getLastPartitionsOrElse(target_tab, null)
|
|
val insert_ds = getLastPartitionsOrElse(s"winhc_ng.inc_${org_prefix}_company", null)
|
|
val insert_ds = getLastPartitionsOrElse(s"winhc_ng.inc_${org_prefix}_company", null)
|
|
|
|
+ if (insert_ds.equals(target_ds)) {
|
|
|
|
+ target_ds = getSecondLastPartitionOrElse(target_tab, null)
|
|
|
|
+ }
|
|
|
|
+ if (target_ds == null) {
|
|
|
|
+ all()
|
|
|
|
+ return
|
|
|
|
+ }
|
|
|
|
|
|
if (target_ds == null) {
|
|
if (target_ds == null) {
|
|
print("target tab is not exists !!!")
|
|
print("target tab is not exists !!!")
|
|
@@ -418,12 +425,16 @@ case class CompanyIndexJob(s: SparkSession,
|
|
val inc_org_tab = s"winhc_ng.inc_${org_prefix}_$tab"
|
|
val inc_org_tab = s"winhc_ng.inc_${org_prefix}_$tab"
|
|
val cols = getColumns(org_tab).intersect(getColumns(inc_org_tab)).map(f => s"cast($f as string) as $f").mkString(",")
|
|
val cols = getColumns(org_tab).intersect(getColumns(inc_org_tab)).map(f => s"cast($f as string) as $f").mkString(",")
|
|
|
|
|
|
|
|
+ val up = cols.contains("update_time") match {
|
|
|
|
+ case true => " DESC,update_time"
|
|
|
|
+ case false => ""
|
|
|
|
+ }
|
|
sql(
|
|
sql(
|
|
s"""
|
|
s"""
|
|
|SELECT *
|
|
|SELECT *
|
|
|FROM (
|
|
|FROM (
|
|
| SELECT *
|
|
| SELECT *
|
|
- | ,ROW_NUMBER() OVER(PARTITION BY ${partition_by.mkString(",")} ORDER BY $order_by DESC ) AS num
|
|
|
|
|
|
+ | ,ROW_NUMBER() OVER(PARTITION BY ${partition_by.mkString(",")} ORDER BY $order_by$up DESC ) AS num
|
|
| FROM (
|
|
| FROM (
|
|
| SELECT ${cols}
|
|
| SELECT ${cols}
|
|
| FROM $org_tab
|
|
| FROM $org_tab
|