My output dataframe has snapshot transaction view. I try to read only “added” rows to output_df since last run but I get zero rows. Is it possible to access newly “added” rows to output_df if it has snapshot view? If not, any pipeline logic suggestion to access those rows?
I want to check if any rows that exist in both input and output dataframes also exist in the newly added data to output_df since the last run.
Transaction view of output_df:
from transforms.api import transform, incremental, Input, Output, configure
from pyspark.sql import types as T
from pyspark.sql import functions as F
schema = T.StructType(
[
T.StructField("wellname", T.StringType()),
T.StructField("start_date", T.TimestampType()),
T.StructField("woe_limit_psia", T.DoubleType()),
T.StructField("end_date", T.TimestampType()),
]
)
@configure(profile=["KUBERNETES_NO_EXECUTORS"])
@incremental(
# require_incremental=True, snapshot_inputs=["input_df"], semantic_version=13
snapshot_inputs=["input_df", "input_df2"], require_incremental=True
)
@transform(
input_df=Input(
"ri.foundry.lava-catalog.dataset.974e09fa-fba6-4867-a7b1-f860ab7a7046"
),
output_df=Output(
"ri.foundry.lava-catalog.dataset.c3407f67-2798-4049-8455-586a025a0b65"
)
)
def incremental_filter(input_df, output_df):
df_current = input_df.dataframe("added")
df_history = output_df.dataframe('previous', schema=schema)
new_output = output_df.dataframe('added', schema=schema)
# I want to check here if any rows exist in input and output dataframes also exist in the newly added data
# to output_df since the last run.
previous_rows = df_current.join(df_history, on=["wellname", "woe_limit_psia"], how="inner")
repeated_rows = previous_rows.join(new_output, on=['wellname', 'woe_limit_psia'], how='left_anti')
# repeated_rows has all data of previous_rows, I think it is because I get 0 rows in new_output.
When an input is snapshoting, all rows of the current View (what you can see in the dataset) are “added rows”, given they were all rewritten by the latest snapshot.
You can’t then differentiate “new rows” from “old rows” given they are all rewritten together.
You can’t as well look at the previous state (the N-1 snapshot).
The only options I see:
- tweak the behavior of your upstream dataset (the one that snapshots) to make it incremental: Having append or update transactions (e.g. in case of update transaction, you will be able to process the update rows as well at the “old rows”)
- Make yourself a diff in your downstream transform: As I guess you have an @incremental() transform context, you can read your current output, which means that you can maybe do a diff between the new upstream snapshot and the current output of your build, to compute what is different/missing and so figure out “what’s new”.