files

env_vars

ModuleNotFoundErrors

August 4th 2020 at 8:56:04pm | prefect.CloudTaskRunner
Unexpected error: NameError("name 'prefect' is not defined")
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 801, in get_task_run_state
    value = timeout_handler(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 188, in timeout_handler
    return fn(*args, **kwargs)
  File "minio.py", line 7, in hello_task
NameError: name 'prefect' is not defined

Traceback (most recent call last):
  File "/home/pedro/prefect/my_flow.py", line 187, in <module>
    flow.register(project_name="myproject", labels=["aws"])
  File "/home/pedro/.venvs/prefect/lib/python3.7/site-packages/prefect/core/flow.py", line 1581, in register
    no_url=no_url,
  File "/home/pedro/.venvs/prefect/lib/python3.7/site-packages/prefect/client/client.py", line 668, in register
    project = self.graphql(query_project).data.project  # type: ignore
  File "/home/pedro/.venvs/prefect/lib/python3.7/site-packages/prefect/client/client.py", line 238, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['project'], 'message': 'field "project" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.project', 'code': 'validation-failed', 'exception': {'message': 'field "project" not found in type: \'query_root\''}}}]

<http://vault.security.banzaicloud.io/vault-addr|vault.security.banzaicloud.io/vault-addr>:
<http://vault.security.banzaicloud.io/vault-role|vault.security.banzaicloud.io/vault-role>:

job_spec

KubernetesJobEnvironment

job_spec.yaml

prefect agent start kubernetes --env NAME=value

PostgresExecute

cloudpickle_deserialization_check

~/.prefect/config.toml

prefect backend server

export PREFECT__CLOUD__API=http://$(curl <http://169.254.169.254/latest/meta-data/public-ipv4):4200/graphql>
export PREFECT__SERVER__UI__GRAPHQL_URL=http://$(curl <http://169.254.169.254/latest/meta-data/public-ipv4):4200/graphql>

Couldn't connect

<http://localhost:4200/graphql>

flow.register()

'project_name' is a required field when registering a flow.

server

project_name

--show-flow-logs

[2020-08-06 23:43:16,058] ERROR - agent | Error while deploying flow: AttributeError("Can't pickle local object 'NpipeHTTPAdapter.__init__.<locals>.<lambda>'")
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Python38\lib\multiprocessing\spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "C:\Python38\lib\multiprocessing\spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
EOFError: Ran out of input

b_flow

a_flow

python
@task
def any_work(a: str) -> str:
    return f'/tmp/{a}'

with Flow('a_flow') as flow:
    name = Parameter('name', default='all')
    result = any_work(name)

python
@task
def any_work(a: str, b: str) -> str:
    return f'{a}/{b}'

with Flow('b_flow') as flow:
    home = Parameter('home', required=True)
    dir = Parameter('dir', default='any')

    result = any_work(home, dir)

python
a_flow = FlowRunTask(flow_name='a_flow', wait=True)
b_flow = FlowRunTask(flow_name='b_flow', wait=True)

with Flow('c_flow') as flow:
    name = Parameter('name', default='all')
    dir = Parameter('dir', default='any')

    a_flow_state = a_flow(parameters={'name': name})

    # this error code, but how i can add parameters in `b_flow` from `a_flow` and dir parameter?
    result = b_flow(
        upstream_tasks=[a_flow_state],
        parameters={
            'home': a_flow_state.result,
            'dir': dir,
        },
    )

set_upstream

set_downstream

with Flow("Thinknum ETL") as flow:

    token = token_task(
        version=version,
        client_id=client_id,
        client_secret=client_secret
    )

    history = history_task(
        dataset_id=dataset_id,
        version=version,
        token=token
    )

    loaded_dates = loaded_dates_task(
        bucket_name=bucket_name,
        dataset_id=dataset_id
    )

    dates_to_load = get_dates_to_load(
        already_loaded_dates=loaded_dates,
        history=history
    )

    datasets_to_load = load_to_gcs_task.map(
        dataset_id=unmapped(dataset_id),
        date=dates_to_load,
        version=unmapped(version),
        token=unmapped(token),
        bucket_name=unmapped(bucket_name)
    )

ModuleNotFoundError

├── alternative_data_pipelines
│   ├── thinknum
|.      ├── __init__.py
│   │   └── thinknum.py
│   └── utils
│       ├── __init__.py
│       ├── logging.py
│       ├── snowflake.py
│       └── utils.py
├── requirements.txt
├── setup.py
└── thinknum_flow.py