https://flyte.org logo
#ask-the-community
Title
# ask-the-community
h

Hema Jayachandran

03/25/2024, 2:36 PM
Hi πŸ‘‹ I am exploring
pyflyte backfill
functionality. We are using a custom function to fetch the input date in our pipelines. The pipeline where I am testing this functionality, runs on current date when no input is provided else picks the passed date. Problem I am facing now is that when I run backfill for the launchplan for 2 days, it is always picking the current date (because of existing logic) for all the subworkflows. While there is no option where we could pass the date via backfill, how can we fully utilise the functionality here ? Any inputs are really appreciated πŸ™‚
s

Samhita Alla

03/26/2024, 8:49 AM
hi! isn't
from-date
or
to-date
suitable for your use case?
h

Hema Jayachandran

03/26/2024, 9:33 AM
Not really. So we use a custom function to get input dates because for majority of our pipelines, it is
current-date - 1
or
current-date
that is required.
Copy code
@task
def get_input_date(date: Optional[str]) -> str:
    date = datetime.now().strftime("%Y-%m-%d") if not date else date
    return date

@task
def input_date_task(date: Optional[str] = None) -> str:
    if not date:
        return (datetime.today() - timedelta(days=1)).strftime("%Y-%m-%d")
    else:
        return date
In this case the behaviour I saw by running the below
Copy code
pyflyte backfill -p project_test -d feature-branch -n test-backfill --execute --from-date 2024-03-23 --to-date 2024-03-25 launch_plan_fi
is a workflow that ran sub workflows for every cron schedule associated to the launch_plan but for each of those executions, it only picked the current date(25th March, 2024) because of our logic. I assume that
from-date
or
to-date
works if our input date picks the
kickoff_time
otherwise its the above behaviour.
we can use
kickoff_time
for our 2 pipelines that runs on current date but won’t work with the rest because of
current-date -1
s

Samhita Alla

03/26/2024, 9:55 AM
sorry i don't understand. where are you using
input_date_task
?
h

Hema Jayachandran

03/26/2024, 9:59 AM
we use it in our workflow, like below
Copy code
@workflow(interruptible=True)
def test_workflow(ecosystem: str, date: Optional[str] = None) -> bool:
    input_date_to_task = get_input_date(date=date)
    path = task_1(date=input_date_to_task)
    ...
we are not passing it as default_inputs because we had problems in the past and we both had a chat about this https://flyte-org.slack.com/archives/CP2HDHKE1/p1689249082278999 πŸ™‚
I am going to test with
kickoff_time
in our pipeline and will post the results here. πŸ™‚
Is there a way we could convert Promise object to datetime ? Our tasks expect date to be in str format so I am trying something like below
Copy code
@workflow(interruptible=True)
def workflow_test(ecosystem: str, kickoff_time: datetime) -> bool:
    # input_date_to_task = get_input_date(date=date)
    input_date_to_task = kickoff_time.strftime("%Y-%m-%d")
but this fails
'Promise' object has no attribute 'strftime'
s

Samhita Alla

03/26/2024, 5:21 PM
oh you cannot materialize promises in a workflow.
you need a task for that.
h

Hema Jayachandran

03/27/2024, 1:37 PM
Thanks Samhita, that worked at task level. gratitude thank you I am still testing out backfill but I can confirm that it works by using the kickoff-time. Will reach out in this thread, if I need more help πŸ™‚
2 Views