|
@@ -34,6 +34,9 @@ case class general_handler(s: SparkSession,
|
|
|
val ads_tab = s"$project.ads_$tn"
|
|
|
val inc_ads_tab = s"$project.inc_ads_$tn"
|
|
|
|
|
|
+ val inter_cols = getColumns(ods_tab).intersect(getColumns(inc_ods_tab)).diff(Seq("rowkey"))
|
|
|
+
|
|
|
+
|
|
|
verify()
|
|
|
reg_udf()
|
|
|
|
|
@@ -97,7 +100,7 @@ case class general_handler(s: SparkSession,
|
|
|
if (md5_fields.isEmpty) {
|
|
|
s"company_id"
|
|
|
} else {
|
|
|
- if (md5_fields.contains("company_id") && job_args.verify_company_id) {
|
|
|
+ if (inter_cols.contains("company_id") && job_args.rowkey_add_company_id) {
|
|
|
s"concat_ws('_',company_id,md5(cleanup(concat_ws('',${md5_fields.mkString(",")}))))"
|
|
|
} else {
|
|
|
s"md5(cleanup(concat_ws('',${md5_fields.mkString(",")})))"
|
|
@@ -153,7 +156,6 @@ case class general_handler(s: SparkSession,
|
|
|
|
|
|
|
|
|
val rowkey_f = get_rowkey_udf()
|
|
|
- val inter_cols = getColumns(ods_tab).intersect(getColumns(inc_ods_tab)).diff(Seq("rowkey"))
|
|
|
|
|
|
|
|
|
val clean_up = get_clean_up()
|