How do I configure a Foundry job to not retry more often than a certain threshold?

user12474509_14945939 0
Viewed 68

I have a Transform job that runs a particular operation I'd like to prevent from occurring more often than some configurable time threshold, that is, it shouldn't run the operation more than once every X minutes.

My code looks like this:

from transforms.api import Input, Output, transform


@transform(
  my_output=Output("/my/output/dataset"),
  my_input=Input("/my/input/dataset")
)
def my_compute_function(my_input, my_output):
  input_df = my_input.dataframe()

  # Do some rate-limited processing...
  my_call()
foundry-code-repositories palantir-foundry
1 Answers

We can use two things here:

  1. An incremental "last updated" output
  2. Abort transaction

We'll store the state of when we last ran inside a second dataset output, and if we detect the timestamp we stored there is less than a certain threshold away, we will abort our run.

This code will look like the following:

from transforms.api import Input, Output, transform, incremental
from pyspark.sql import types as T
from datetime import datetime, timedelta

MIN_TIME_ELAPSED = timedelta(minutes=30)

RUN_RECORD_SCHEMA = StructType([ \
  StructField("last_run", T.TimestampType(), False),
])


def maybe_get_run_rows(run_record):
  last_run_df = run_record.dataframe("previous", schema=RUN_RECORD_SCHEMA)
  last_run_rows = last_run_df.collect()
  return last_run_rows


def within_run_window(run_record_rows):
  current_run_time = datetime.now()
  last_run_time = run_record_rows[0][0]
  return (current_run_time - last_run_time) > MIN_TIME_ELAPSED:
  

def record_run(run_record, ctx):
  # Record that we ran
  new_run_record_df = ctx.spark_session.createDataFrame(
    [[current_run_time]],
    schema=RUN_RECORD_SCHEMA
  )
  run_record.set_mode("replace")
  run_record.write_dataframe(new_run_record_df)



@incremental(semantic_version=1)
@transform(
  my_output=Output("/my/output/dataset"),
  run_record=Output("/my/output/run_record"),
  my_input=Input("/my/input/dataset")
)
def my_compute_function(my_input, run_record, my_output, ctx):
  input_df = my_input.dataframe()

  run_record_rows = maybe_get_run_rows(run_record)

  if len(run_record_rows) == 0:
    # First run ever

    # Do some rate-limited processing...
    my_call()
    record_run(run_record, ctx)

  else:
    # Not first run ever

    if not within_run_window(run_record_rows):
      # Built too quickly, do nothing

      my_output.abort()
      return

    else:
      # Proceed as normal

      my_call()
      record_run(run_record, ctx)

You can now configure the MIN_TIME_ELAPSED argument by passing the minimum time elapsed amount you'd like, per the timedelta docs

If for some reason you wanted to only run one time ever, then you could slightly modify your compute function to instead be:

@incremental(semantic_version=1)
@transform(
  my_output=Output("/my/output/dataset"),
  run_record=Output("/my/output/run_record"),
  my_input=Input("/my/input/dataset")
)
def my_compute_function(my_input, run_record, my_output, ctx):
  input_df = my_input.dataframe()

  run_record_rows = maybe_get_run_rows(run_record)

  if len(run_record_rows) == 0:
    # First run ever

    # Do some rate-limited processing...
    my_call()
    record_run(run_record, ctx)

  else:
    # Not first run ever

    my_output.abort()
    return
user12474509_149459390
Related