UnsupportedOperationException: dataType about spline-spark-agent

[wajda]

Could you please provide a Spark app snippet that reproduces the issue?

[wajda]

What Spark version are you using?

[arijit999]

Spark Version is 2.4.4

PySpark 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)

[cerveada]

• WindowSpecDefinition.dataType always throws.

• This expression is most probably created when using column.over(...) function.

• Since WindowSpecDefinition extends Unevaluable that doesn't suppose to live past analysis or optimization, the solution may be to ignore it.

[cerveada]

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