Open for discussion about Prefect's Core engine: https://github.com/PrefectHQ/prefect
pickle's error message was better than cloudpickle's
Sweet. Thanks @cicdw . I'm sure my issue is quite simple. I started with the write_tweets_to_airtable.py example, but stripped it down to the bare bones (almost hello world). Here it is.
import prefect
from prefect import Flow, Parameter, task
@task
def say_hello(name):
print(f"Hello, {name}!")
def main():
with Flow("My First Flow") as flow:
twitter_user = Parameter("twitter_user")
# tweet_id = Parameter("tweet_id", required=False)
say_hello(twitter_user)
flow_context = dict()
with prefect.context(flow_context):
# run the flow
state = flow.run(parameters=dict(
twitter_user="Josh"
# tweet_id="1046873748559814656"
))
if __name__ == '__main__':
main()
Now this works, but if I add uncomment both lines related to the tweet_id Parameter, I get the following error,
Traceback (most recent call last):
File "first-flow.py", line 30, in <module>
main()
File "first-flow.py", line 25, in main
tweet_id="1046873748559814656"
File "/Users/josh/Envs/prefect/lib/python3.6/site-packages/prefect/core/flow.py", line 976, in run
fmt_params
ValueError: Flow.run received the following unexpected parameters: tweet_id
so you could add it to your flow via:
flow.add_task(tweet_id)
and then your example should run, without updating
say_hello
anytime! Outside of the Getting Started Docs and Concept docs (which it sounds like you've used), there isn't another good document for explaining Task behavior like this. This is something that has come up before though, and so I've prioritized writing one that would cover this and other things
I hope to have it out sometime next week
the task flow logic seems pretty easy to write and could easily be run from start to finish within a docker container that runs on a scheduled basis
Hi @braunreyes - you can definitely run your flow inside an ephemeral container; most people want the full power of the Cloud product because it includes a live-updating database of all task states, which then drives the web interface and also makes the system more robust generally (in case the process running the flow dies for anyone reason, for example)
Running your Flow inside a container should work fine though, it just might not always be transparent to debug when things go wrong. However, given the limited availability of Cloud at this moment I think it's a reasonable way to go for the time being
i think its fine if they just have logs for what failed
@cicdw been a while since I was here, lots of life things! to reply to your question: I believe the snapshots should live locally much of the time, but sometimes in the cloud. I think in the general case I'm a little less concerned about build times (which is how I think about the benefits of cloud storage) and more about consistency and reproducibility which can all be handled locally
hey everyone! In case you haven't seen, by popular request we are going to migrate our public discussions to Slack for better history / organization: https://t.co/UyS12hy3mO
I'll leave the gitter open for a while to reroute people there
@/all ^^
@cicdw Regarding state and metadata (such as a log of Task runs), you mentioned a while back that Prefect Cloud will have this setup by default. If we wanted to roll our own DB, how custom is this kind of setup and is there any Airflow-like equivalent to setting up your "metadata db"?
Ah it expired - my apologies! Try this one: https://join.slack.com/t/prefect-public/shared_invite/enQtNzE5OTU3OTQwNzc1LTQ5M2FkZmQzZjI0ODg1ZTBmOTc0ZjVjYWFjMWExZDAyYzBmYjVmMTE1NTQ1Y2IxZTllOTc4MmI3NzYxMDlhYWU
Hi @jamesn43543455_twitter - use a
LocalExecutor (https://docs.prefect.io/api/unreleased/engine/executors.html#localexecutor)
also FYI we don't monitor this channel much anymore - for faster responses I'd recommend joining our Slack channel: https://join.slack.com/t/prefect-public/shared_invite/enQtNzE5OTU3OTQwNzc1LTQ5M2FkZmQzZjI0ODg1ZTBmOTc0ZjVjYWFjMWExZDAyYzBmYjVmMTE1NTQ1Y2IxZTllOTc4MmI3NzYxMDlhYWU
Hi @mmccarty ! Sorry for the delayed response, we don't check the gitter channel as frequently anymore (we migrated to slack). When you say "run a Flow asynchronously", I assume you are referring to the individual tasks running asynchronously -- for this we recommend using the
DaskExecutor which supports asynchronous execution (some relevant docs here: https://docs.prefect.io/guide/tutorials/dask-cluster.html), but let me know if I misunderstood your question!
hi all, im new to Prefect and workflow management systems in general. i have a quick question: what is the typical recommendation around the scheduling logic and business logic; should it be kept together or separate? for example, should the tasks being called be ran in a separate service?
Hi everyone, are there any plans to support python-trio in DaskExecutor, either natively or using AnyIO? (https://github.com/agronholm/anyio)
I would be open to contributing but I would need some collaboration, it's not something I think I can pull off on my own.
Hi! I am writing a custom notifier and want to output the input-arguments of my task. For example, I want to send to slack p_table_name
@task
def make_query_for_columns(p_table_name) -> str:
my handler:
def my_state_handler(obj: TrackedObjectType, old_state: "prefect.engine.state.State",
new_state: "prefect.engine.state.State"):
Can you tell us how to do this?