Comments (5)
Could you please provide a Spark app snippet that reproduces the issue?
from spline-spark-agent.
What Spark version are you using?
from spline-spark-agent.
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)
from spline-spark-agent.
-
WindowSpecDefinition.dataType always throws.
-
This expression is most probably created when using
column.over(...)
function. -
Since
WindowSpecDefinition
extendsUnevaluable
that doesn't suppose to live past analysis or optimization, the solution may be to ignore it.
from spline-spark-agent.
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
from spline-spark-agent.
Related Issues (20)
- Datahub integration reference link HOT 1
- Getting error while connecting tryng to use Kafka dispatcher HOT 1
- Initialization failure handling control HOT 2
- Cant track org.apache.hadoop.fs.rename
- use splineAgent as a Jar file HOT 12
- Spline Agent required for spark 3.4 version and Scala 2.12 HOT 2
- Facing issues while sending Databricks Lineage (Spline) to kafka HOT 1
- Support for spark 3.5 HOT 4
- Does spline solution support snowpark
- Support of setting the arangoDB name on the configuration HOT 7
- Need to deploy on Synapse environment in order to track lineage for Synapse notebooks HOT 1
- Unity Catalog Support to get the Notebook details
- Support for AWS Glue DynamicFrame HOT 2
- Error when running spark application with agent-core as queryExecutionListeners in java 17
- Spline Support of Statment Create table as select HOT 11
- Using Spline in Spark Scala project HOT 1
- Can Spline support lineage for AWS glue Spark dynamic frames? HOT 1
- Spline doesn't track some in memory operations in pyspark HOT 5
- Customized Listener HOT 1
- Databricks - Support of Delta live tables
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from spline-spark-agent.