https://flyte.org logo
Join SlackCommunities
Powered by
Anyone know how to create a launch plan programmat...
# flyte-support
e
Anyone know how to create a launch plan programmatically in Flyte? I've been trying to run this code:
Copy code
LaunchPlan.get_or_create(
    name="my_launchplan",
    workflow=wf,
    schedule=CronSchedule(
        schedule="0 6 * * MON#1",  # UTC, see <https://docs.flyte.org/en/latest/concepts/schedules.html#cron-expression-table>
    )
)
but running it directly with Python doesn't seem to do anything. I've tried 
pyflyte launchplan
and 
pyflyte run --remote
, but neither seems to support this use case. Any help would be much appreciated! 🙏 (I'm running the example from here: https://github.com/flyteorg/flytesnacks/blob/0ec8388759d34566a0ffc0c3c2d7443fd4a3a46f/examples/basics/basics/launch_plan.py)
f
you can , we should write a method in remote.py to do this (its currently behind raw_register) sadly
you can use 
raw_register
e
Hi @freezing-airport-6809 Thanks for answering👋 I'm currently working on this PR and am trying to create a launch plan with an intentionally invalid cron string to validate the recent changes. I attempted the following setup:
Copy code
python
@task
def mean(values: List[float]) -> float:
    return sum(values) / len(values)

@task
def standard_deviation(values: List[float], mu: float) -> float:
    variance = sum([(x - mu) ** 2 for x in values])
    return sqrt(variance)

@task
def standard_scale(values: List[float], mu: float, sigma: float) -> List[float]:
    return [(x - mu) / sigma for x in values]

@workflow
def standard_scale_workflow(values: List[float]) -> List[float]:
    mu = mean(values=values)
    sigma = standard_deviation(values=values, mu=mu)
    return standard_scale(values=values, mu=mu, sigma=sigma)

if __name__ == "__main__":
    remote = FlyteRemote(config=Config.auto())

    standard_scale_launch_plan = LaunchPlan.get_or_create(
        workflow=standard_scale_workflow,
        name="standard_scale_lp_haha",
        schedule=CronSchedule(schedule="* * * * MON#1"),
    )

    remote_standard_scale_workflow: FlyteWorkflow = remote.register_workflow(
        standard_scale_workflow,
    )

    remote.register_launch_plan(standard_scale_launch_plan, version="v1", project="flytesnacks", domain="development")
Unfortunately, this approach didn’t work as expected, and I suspect there might be an issue with my process. I also explored the 
raw_register
method you mentioned, but I couldn’t find much documentation on using it specifically to register a launch plan. If you have any insights or additional guidance, I would greatly appreciate it! Thank you very much.
f
yes will try to add the method once i get a chance
aah i see its there
i think this seems like a bug in the tracker
cc @witty-action-93503 can you please take a look, the above script fails with
Copy code
╭───────────────────── Traceback (most recent call last) ──────────────────────╮
│ /Users/ketanumare/src/ketan-examples/extras/register_main.py:38 in <module>  │
│                                                                              │
│ ❱ 38 │   remote_standard_scale_workflow = remote.register_workflow(          │
│                                                                              │
│ /Users/ketanumare/src/flytekit/flytekit/remote/remote.py:939 in              │
│ register_workflow                                                            │
│                                                                              │
│ ❱  939 │   │   │   _, _, _, module_file = extract_task_module(entity)        │
│                                                                              │
│ /Users/ketanumare/src/flytekit/flytekit/core/tracker.py:354 in               │
│ extract_task_module                                                          │
│                                                                              │
│ ❱ 354 │   │   │   name = f.lhs                                               │
│                                                                              │
│ /Users/ketanumare/src/flytekit/flytekit/core/tracker.py:120 in lhs           │
│                                                                              │
│ ❱ 120 │   │   return self.find_lhs()                                         │
│                                                                              │
│ /Users/ketanumare/src/flytekit/flytekit/core/tracker.py:172 in find_lhs      │
│                                                                              │
│ ❱ 172 │   │   raise _system_exceptions.FlyteSystemException(f"Error looking  │
╰──────────────────────────────────────────────────────────────────────────────╯
FlyteSystemException: SYSTEM:Unknown: error=Error looking for LHS in 
register_main
it seems we broke the 
__main__
handling?
w
Sure, I’ll give it a look when I have time
f
thank you
w
I think 
lhs
is only used for imported instances, which is not really ideal (maybe we can fix that in the future?). One way to bypass this error is to move all wf & task definitions into another file and import them into your main module.
26 Views
Open in Slack
PreviousNext
Powered by