UnsupportedOperationException: dataType #28
Comments
|
Could you please provide a Spark app snippet that reproduces the issue? |
|
What Spark version are you using? |
Spark Version is 2.4.4PySpark Code Snippet#import statements
def rm_detl(
cgm_dtl: DataFrame, op: DataFrame, rm_batch_qr7_disc: DataFrame, RPT_MNTH: Date
) -> DataFrame:
cgm_dtl = cgm_dtl.select(
fx.col("PGM_ID"),
fx.col("UPDATED_ON"),
fx.col("CREATED_ON"),
fx.col("STAT_RSN_CODE"),
fx.col("STAT_CODE"),
fx.col("BEG_DATE"),
fx.col("END_DATE"),
fx.col("CREATED_BY"),
).alias("cgm_dtl")
op = op.select(
fx.col("PID"), fx.col("SERIAL_NUM_IDENTIF"), fx.col("CASE_NAME")
).alias("op")
rm_batch_qr7_disc = rm_batch_qr7_disc.select(
fx.col("PID"), fx.col("STAFF_ID")
).alias("rmbqr7")
pdh_qr7_join = (
rm_batch_qr7_disc.join(
cgm_dtl, fx.col("rmbqr7.PID") == fx.col("cgm_dtl.PGM_ID")
)
.select(fx.col("STAFF_ID"))
.alias("pdh_qr7_join")
)
qry_pgmdtl_op_1 = (
(cgm_dtl.join(op, fx.col("cgm_dtl.PGM_ID") == fx.col("op.PID")))
.where(
(
fx.to_date(fx.col("cgm_dtl.BEG_DATE"))
<= fx.add_months(fx.to_date(fx.lit(RPT_MNTH)), -1)
)
& (
fx.to_date(fx.col("cgm_dtl.END_DATE"))
> fx.add_months(fx.to_date(fx.lit(RPT_MNTH)), -1)
)
& (
fx.to_date(fx.col("cgm_dtl.CREATED_ON"))
< fx.to_date(fx.lit(RPT_MNTH))
)
)
.select(
fx.col("cgm_dtl.PGM_ID"),
fx.col("cgm_dtl.UPDATED_ON"),
fx.col("cgm_dtl.CREATED_ON"),
fx.col("cgm_dtl.STAT_RSN_CODE"),
fx.col("cgm_dtl.STAT_CODE"),
fx.col("cgm_dtl.BEG_DATE"),
fx.col("cgm_dtl.END_DATE"),
fx.col("cgm_dtl.CREATED_BY"),
fx.col("op.PID"),
fx.col("op.SERIAL_NUM_IDENTIF"),
fx.col("op.CASE_NAME"),
)
.alias("qry_pgmdtl_op_1")
)
qry_pgmdtl_op_2 = (
cgm_dtl.join(op, fx.col("cgm_dtl.PGM_ID") == fx.col("op.PID"))
.join(
pdh_qr7_join,
fx.col("cgm_dtl.CREATED_BY") == fx.col("pdh_qr7_join.STAFF_ID"),
)
.where(
(
(fx.col("cgm_dtl.BEG_DATE") == fx.to_date(fx.lit(RPT_MNTH)))
& (
fx.trunc(fx.col("cgm_dtl.CREATED_ON"), "mon")
== fx.trunc(fx.to_date(fx.lit(RPT_MNTH)), "mon")
)
& (fx.col("cgm_dtl.STAT_CODE") == fx.lit("DS"))
& (
fx.col("cgm_dtl.STAT_RSN_CODE").isin(
["01", "02", "03", "SC", "SD", "SB"]
)
)
)
)
.select(
fx.col("cgm_dtl.PGM_ID"),
fx.col("cgm_dtl.UPDATED_ON"),
fx.col("cgm_dtl.CREATED_ON"),
fx.col("cgm_dtl.STAT_RSN_CODE"),
fx.col("cgm_dtl.STAT_CODE"),
fx.col("cgm_dtl.BEG_DATE"),
fx.col("cgm_dtl.END_DATE"),
fx.col("cgm_dtl.CREATED_BY"),
fx.col("op.PID"),
fx.col("op.SERIAL_NUM_IDENTIF"),
fx.col("op.CASE_NAME"),
)
.alias("qry_pgmdtl_op_2")
)
uni_pgmdtl_op_1_and_2 = qry_pgmdtl_op_1.union(qry_pgmdtl_op_2).alias(
"uni_pgmdtl_op_1_and_2"
)
w_mx_beg_dt_prtn_on_pid = Window.partitionBy(uni_pgmdtl_op_1_and_2.PID)
a = (
uni_pgmdtl_op_1_and_2.withColumn(
"MX_BEG_DATE",
fx.max(fx.col("uni_pgmdtl_op_1_and_2.BEG_DATE")).over(
w_mx_beg_dt_prtn_on_pid
),
)
.distinct()
.alias("a")
)
w_mx_updt_dt_prtn_on_pid = Window.partitionBy(a.PID)
b = (
a.where(fx.col("a.BEG_DATE") == fx.col("a.MX_BEG_DATE"))
.withColumn(
"MX_UPDATED", fx.max(fx.col("a.UPDATED_ON")).over(w_mx_updt_dt_prtn_on_pid)
)
.alias("b")
)
rm_detl_df = (
b.where(fx.col("b.UPDATED_ON") == fx.col("b.MX_UPDATED")).select(
fx.col("b.SERIAL_NUM_IDENTIF").alias("SERIAL_NUM_IDENTIF"),
fx.col("b.CASE_NAME").alias("CASE_NAME"),
fx.lit(None).alias("TS").cast(TimestampType()).alias("TS"),
fx.col("b.PID").alias("PID"),
fx.col("b.UPDATED_ON").alias("UPDATED_ON"),
fx.date_trunc("day", fx.col("b.CREATED_ON")).alias("CREATED_ON"),
fx.col("b.STAT_RSN_CODE").alias("STAT_RSN_CODE"),
fx.col("b.STAT_CODE").alias("STAT_CODE"),
fx.col("b.BEG_DATE").alias("BEG_DATE"),
fx.col("b.END_DATE").alias("END_DATE"),
fx.col("b.CREATED_BY").alias("CREATED_BY"),
fx.col("b.MX_BEG_DATE").alias("MX_BEG_DATE"),
fx.col("b.MX_UPDATED").alias("MX_UPDATED"),
)
).alias("rm_detl_df")
return rm_detl_df
class KmDtl(SparkJob):
def __init__(
self, Date:str, database=dl.ZZZ, bucket=dl.DEFAULT_BUCKET
):
super().__init__(database=database, bucket=bucket, Date=Date)
self.database = database
self.bucket = bucket
self.RPT_MNTH = datetime.date(datetime.strptime(Date, "%Y-%m-%d"))
def load_all_data(self):
self.CASE = dl.load_from_datalake(self.database, dl.CASE, bucket=self.bucket)
self.CGM = dl.load_from_datalake(self.database, dl.CGM, bucket=self.bucket)
self.CGM_DTL = dl.load_from_datalake(self.database, dl.CGM_DTL, bucket=self.bucket)
self.STAFF = dl.load_from_datalake(self.database, dl.STAFF, bucket=self.bucket)
self.OP = open_pgms(
CASE=self.CASE,
CGM=self.CGM,
CGM_DTL=self.CGM_DTL,
RPT_MNTH=self.RPT_MNTH,
)
self.RM_BATCH_QR7_DISC = rm_batch_qr7_disc(
STAFF=self.STAFF, CGM_DTL=self.CGM_DTL, RPT_MNTH=self.RPT_MNTH
)
def run(self):
spark = SparkSession.builder.appName("KmDtl").getOrCreate()
self.load_all_data()
rm_detl_df = rm_detl(
cgm_dtl=self.CGM_DTL,
op=self.OP,
rm_batch_qr7_disc=self.RM_BATCH_QR7_DISC,
RPT_MNTH=self.RPT_MNTH,
)
dl.save_to_datalake(rm_detl_df, dl.RR2255CW, dl.RM_DETL, bucket=self.bucket) |
|
|
We will release fix for 0.5.x that will allow agent to gather the data properly. Attribute lineage feature will not take into account the information from windowExpressions - this will be fixed in spline 0.6 as AbsaOSS/spline#668 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
The text was updated successfully, but these errors were encountered: