prefect-community #prefect-community

What's the best way to set these dependencies in Prefect. I notice that set_downstream doesn't accept a list of tasks. Is there any efficient way to do this in Prefect? Also does the new Prefect UI mean I don't need docker to run it? Thanks

Vikram Iyer

08/10/2020, 6:05 AM

If I restart the apollo, ui and agent containers, I lose the flows. Is there any way to persist them?

Vikram Iyer

08/10/2020, 6:05 AM

Is there a volume on the host that I can map to docker container so that it picks up the flows automatically after server restart?

emmanuel

08/10/2020, 7:20 AM

Hello there, coming from an Airflow background, I was wondering if there is already something built-in like an ExternalTaskSensor to declare dependencies between different flows like in Airflow?

Lewis Bails

08/10/2020, 8:26 AM

Hello prefectors! I'm trying to get the DaskKubernetesEnvironment to work and I'm running into what I think is a Docker storage problem. I can build the image outside of the Prefect process, so I don't think my issue is with the dockerfile.

Copy code

flow.environment = DaskKubernetesEnvironment(min_workers=1s, max_workers=5)
flow.storage = Docker(dockerfile='./dockerfile', env_vars={'known_hosts': #known_hosts, 'ssh_prv_key': #prv_key})
flow.register(project_name='Automated training')

Copy code

[2020-08-10 08:07:11] INFO - prefect.Docker | Building the flow's Docker storage...
Traceback (most recent call last):
  File "/home/leb/anaconda3/envs/ai-pipelines/lib/python3.7/site-packages/docker/api/client.py", line 261, in _raise_for_status
    response.raise_for_status()
  File "/home/leb/.local/lib/python3.7/site-packages/requests/models.py", line 941, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http+docker://localhost/v1.40/build?t=auto-dask-kube-cloud%3A2020-08-10t08-07-11-222981-00-00&q=False&nocache=False&rm=False&forcerm=True&pull=False&dockerfile=.%2Ftmpez2venih%2FDockerfile>

Can anyone help me out?

Roy

08/10/2020, 11:00 AM

Hello! When I

merge

two task results together in a single one, the merge is supposed to return the first "real" result. What is considered a "real" result? It seems that only

None

is considered a non-real result. I would have expected that

NoResult

results would also be skipped, but this is not the case. Is this intended? Example in thread.

Amit

08/10/2020, 11:24 AM

Hi Team, How can make the prefect to log the messages from the default python logger to log to cloud, so for example If I get the logger via:

Copy code

logger = logging.getLogger(__name__)

I have this in code everywhere and I don't want replace it with

from prefect.utilities.logging import get_logger

everywhere, is there a way to send the default logger messages to cloud?

Jim Klassen

08/10/2020, 2:23 PM

Hi everyone.

Jim Klassen

08/10/2020, 2:25 PM

I am getting a heartbeat error as the last message from my flow, I believe everything worked correctly. Does anyone know why?

Marwan Sarieddine

08/10/2020, 3:44 PM

Hi folks - anyone faced this error before ?

Copy code

$ kubectl logs pod/prefect-job-d45c5eda-cxn9x -n prefect
Usage: prefect execute [OPTIONS] COMMAND [ARGS]...
Try 'prefect execute -h' for help.

Error: No such command 'flow-run'.

bral

08/10/2020, 3:47 PM

Good evening ! I faced with problem : on same server started prefect , and local agent. If i try go to http//serverip:8080 from remote machine i do not see any flow, because ui trying to connect to httplocalhost4200/graphql. Which options i should change ?

Philip MacMenamin

08/10/2020, 4:23 PM

Is there a standard way to pass a

list

as a

Parameter

? (As opposed to a

str

- or do you just pass a string and split?)

Marwan Sarieddine

08/10/2020, 5:27 PM

I am facing a cloud UI bug on chrome- when the flow is too large -and I visit the schematic tab, the app (cloud.prefect.io) crashes - and opening a new tab to go to cloud.prefect.io doesnt solve things, I have to open up an incognito session to clear browser session storage and resolve things

Marwan Sarieddine

08/10/2020, 5:49 PM

Is there a way using the UI logs tab to open the task that caused an error ?

Philip MacMenamin

08/10/2020, 6:36 PM

Copy code

with Flow("STL Flow") as f:
    files = Parameter("files")

f.run(files='fname.txt')

Riley Hun

08/10/2020, 8:57 PM

Not sure if this is a silly question - sorry if it is - but if you store your secrets in a docker file, what's difference between using the EnvVarSecret task and using os.environ to pull in the secrets?

Riley Hun

08/11/2020, 4:49 AM

Hello again, Was wondering if there's a way to quickly test a flow using docker storage without actually registering it to the UI. Thanks so much in advance. Cheers, Riley

Lewis Bails

08/11/2020, 6:05 AM

Hi there, I have a task that is being wrongfully executed multiple times despite only showing up once in the flow. Has anyone had this issue before? Edit: I'm using Prefect 0.13.1, with Dask 2.22.0, LocalEnvironment with DaskExecutor

emre

08/11/2020, 7:52 AM

Hi folks, In prefect

0.13.1

, I can’t get the

slack_notifier

to work. This used to work in

0.12.1

. Here is a minimal example of a flow:

Copy code

from prefect import Flow, task, context
from prefect.utilities.notifications import slack_notifier

@task
def log_run_id():
    context.get("logger").warning(context.get("flow_run_id"))


@task(state_handlers=[slack_notifier])
def get_1():
    return 1


with Flow("slack_example") as flow:
    log = log_run_id()
    result = get_1(upstream_tasks=[log])


flow.run()

Apparently

slack_notifier

attempts to communicate with a backend server, but I am not going to use one. More info in thread.

bral

08/11/2020, 9:35 AM

After upgrading to 0.13.1 version tasks dont scheduled with local agent. I noticed that docker-compose.yml changed and sheduler.py missed from src dir. So agent does not start task

emmanuel

08/11/2020, 11:31 AM

is there a way to run the OSS server and the UI in a secured way, that is not exposing the apollo endpoint publicly? or protecting it?

emmanuel

08/11/2020, 2:29 PM

PREFECT_SERVER__GRAPHQL_URL

the right env variable for the UI docker image? it seems to be the one used in the docker compose file but when ran outside of docker (so not on localhost), it doesn’t seem to be working 😕

Richard Hughes

08/11/2020, 2:55 PM

Hi, is there a way to stop all running flows? If I have a bunch of running flows that are hung, how can I stop them quickly?

Chris Martin

08/11/2020, 2:59 PM

Hi- I'm trying to set an upstream task on

task.map

, but I can't get it to work and wonder if this is possible Here's some example code:

Copy code

from prefect import task, Flow, Task

@task
def task1():
    pass

@task
def task2():
    return ["a", "b", "c"]

@task
def task3(x):
    pass


with Flow("test") as flow:
    t1 = task1()
    params = task2()
    task3.map(params, upstream_tasks=[t1])

I can register this flow fine and the dag looks coorrect (see attachment), but when I run,

task3

fails with "No upstream states can be mapped over".Am I doing something wrong here?

Akshay Verma

08/11/2020, 3:12 PM

Hi, Maybe a repeated/naive question. With the latest release, is it possible to run Prefect for data pipelines completely within a VPN without connecting to an external services? We have to refactor some custom scripts and we are considering between Airflow and Prefect. With Prefect, we are not clear that we can use it independent of the Prefect Cloud.

emmanuel

08/11/2020, 3:53 PM

silly question but is

apollo

OSS as well?

emmanuel

08/11/2020, 4:05 PM

@nicholas so when I see stuff like this https://github.com/PrefectHQ/server/blob/master/services/apollo/src/executors.js#L5 does that mean again that it’s injected at build time?

Hannah Amundson

08/11/2020, 4:54 PM

i have a question about logging best practices. so i understand that to get the prefect logger we need to do

Copy code

prefect.context.get("logger")

How do we get that logger inside of each of our classes/functions that are being called within tasks? Is the best practice to just pass it in as a parameter?

Riley Hun

08/11/2020, 6:48 PM

Hello, I am a beginner at prefect pipeline orchestration. This is the first time I've attempted to dockerize my prefect pipeline and register it to local UI server. After submitting my flow, it seems to be stuck with a backlog of "late runs".

Marwan Sarieddine

08/11/2020, 9:36 PM

Hi folks, I am facing this error - whenever my flow involves spinning up a considerable number of workers (more than 100 dask workers) - note the same flow runs fine when I set a smaller number of workers… (the error is invoked after the flow goes into Running state, but before any tasks are run)

Copy code

Unexpected error: TimeoutError()
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 410, in get_flow_run_state
    with executor.start():
  File "/usr/local/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/executors/dask.py", line 239, in start
    with Client(self.address, **self.client_kwargs) as client:
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 744, in __init__
    self.start(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 949, in start
    sync(self.loop, self._start, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 339, in sync
    raise exc.with_traceback(tb)
  File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 323, in f
    result[0] = yield future
  File "/usr/local/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1046, in _start
    await self._ensure_connected(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1108, in _ensure_connected
    await asyncio.wait_for(self._update_scheduler_info(), timeout)
  File "/usr/local/lib/python3.7/asyncio/tasks.py", line 449, in wait_for
    raise futures.TimeoutError()
concurrent.futures._base.TimeoutError

it seems to me the solution to avoid this timeout is to add a task that waits for the workers to be ready - anyone else run into this ? (I am using a DaskKubernetesEnvironment for execution on AWS EKS)