dask-resource

tags

@task

running_server = "name_of_remote_server"
resource_tag = "dask-resource:{}=1".format(running_server)
@task(log_stdout=True, state_handlers=[email_on_failure], tags=[resource_tag])
def test_task():
    print(1/0)

dask-worker tcp://<address of local dask-scheduler>:8786 --nprocs 4 --nthreads 1 --worker-port xxx --resources "name_of_remote_server=1"

ModuleNotFoundError: No module named 'prefect'

[2020-11-24 17:02:05+0800] ERROR - prefect.CloudTaskRunner | Failed to set task state with error: ConnectionError(MaxRetryError("HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7ff6099fcf10>: Failed to establish a new connection: [Errno 111] Connection refused'))"))
Traceback (most recent call last):
  File "/anaconda3_a/envs/py37/lib/python3.7/site-packages/urllib3/connection.py", line 160, in _new_conn
    (self._dns_host, self.port), self.timeout, **extra_kw
  File "/anaconda3_a/envs/py37/lib/python3.7/site-packages/urllib3/util/connection.py", line 84, in create_connection
    raise err
  File "/anaconda3_a/envs/py37/lib/python3.7/site-packages/urllib3/util/connection.py", line 74, in create_connection
    sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused
...
2020-11-24 17:02:05+0800] INFO - prefect.CloudTaskRunner | Task 'test_task': Finished task run for task with final state: 'ClientFailed'
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Failed to write log with error: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7ff6097590d0>: Failed to establish a new connection: [Errno 111] Connection refused'))
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Failed to write log with error: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7ff6097590d0>: Failed to establish a new connection: [Errno 111] Connection refused'))
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Failed to write log with error: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7ff6097590d0>: Failed to establish a new connection: [Errno 111] Connection refused'))
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Unable to write logs to Prefect Cloud
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Unable to write logs to Prefect Cloud
[2020-11-24 09:02:10+0000] CRITICAL - CloudHandler | Unable to write logs to Prefect Cloud

prefect.utilities.exceptions.ClientError: 400 Client Error: Bad Request for url: http://<Prefect Cloud IP>:4200

This is likely caused by a poorly formatted GraphQL query or mutation. GraphQL sent:

query {
  query { auth_info { api_token_scope } }
}
variables {
  null
}

pending

@task

flow.run(executor=LocalDaskExecutor(scheduler="threads", num_workers=4))

flow.storage = Docker(....)
flow.run_config = DockerRun()
flow.executor = LocalDaskExecutor(scheduler="threads", num_workers=4)
flow.register(project_name=... , build=False)

run_config

dbt_task = DbtShellTask(
  ...
)
with Flow("dbt-flow") as flow:
    dbt_command_string = Parameter('dbt_command_string', default = 'dbt run', required = True)
    dbt_commands = dbt_command_string.split(',')
    for command in dbt_commands:
      dbt_task(
          command=dbt_command,
          ...
      )

map

Unexpected error while running flow: KeyError('Task slug foo-3 not found in the current Flow; this is usually caused by changing the Flow without reregistering it with the Prefect API.')

0.13.18

FROM python:3.8
RUN python -m pip install sqlalchemy prefect

from prefect import Flow, task
from prefect.environments.storage import Docker
import sqlalchemy


@task
def some_task():
    print(sqlalchemy.__version__)


with Flow("Some Flow") as flow:
    some_task()

if __name__ == "__main__":
    flow.storage = Docker(
        registry_url="<http://docker.io|docker.io>",
        image_name="storage_image",
        image_tag="latest",
        base_image="test-base",
        local_image=True,
    )
    flow.register(project_name="Development")

#!/usr/bin/env bash
set -x

docker build . -t test-base

#: Approach 1
python main.py

#: Approach 2
docker run -v /var/run/docker.sock:/var/run/docker.sock -v ~/.docker:/root/.docker -e PREFECT__CLOUD__AUTH_TOKEN="$PREFECT__CLOUD__AUTH_TOKEN" -v $(pwd)/main.py:/main.py test-base bash -c 'python /main.py'

start_date

scheduled_start_time

Parameter

default

Parameter

validator

from prefect import Flow, task
from prefect.environments.storage import Docker
import sqlalchemy


@task
def some_task():
    print(sqlalchemy.__version__)


with Flow("test") as flow:
    some_task()

if __name__ == "__main__":
    flow.storage = Docker(
        registry_url="<http://docker.io/atheon|docker.io/atheon>",
        image_name="prefect-test",
        base_image="atheon/prefect-test:base",
    )
    flow.register(project_name="Development")

---> Running in 26ae91f84100
Beginning health checks...
System Version check: OK
Cloudpickle serialization check: OK
Result check: OK
Environment dependency check: OK
All health checks passed.
Removing intermediate container 26ae91f84100
 ---> a63da9840884
Traceback (most recent call last):
  File "register.py", line 60, in <module>
    storage.build()
  File "/home/karteek/anaconda3/envs/prefect/lib/python3.6/site-packages/prefect/environments/storage/docker.py", line 359, in build
    self._build_image(push=push)
  File "/home/karteek/anaconda3/envs/prefect/lib/python3.6/site-packages/prefect/environments/storage/docker.py", line 423, in _build_image
    self._parse_generator_output(output)
  File "/home/karteek/anaconda3/envs/prefect/lib/python3.6/site-packages/prefect/environments/storage/docker.py", line 661, in _parse_generator_output
    or parsed.get("errorDetail", {}).get("message")
AttributeError: 'NoneType' object has no attribute 'strip'

flow.run()

flow.register()

Unexpected error: TypeError("can't pickle _thread.lock objects")

flow.run()

flow.register()

register()