I am trying to work out how to work with relational database sources in Prefect. Does anyone have an example? The usual process is that I have a bunch of tables that needs to be extracted and later loaded somewhere. I have tried the postgres-task. It works, but how would I structure my code. It is not always great to read 100M rows into memory by using

fetch='all'

. If I want to read 5000 rows, store them somewhere and then read another 5000 rows. How would i proceed with prefect?

Hi, We have a set of workflows that we are in process of prefectifying, but have stumbled on a problem. We are processing weather data, and the preferred python libraries for handling meteorological data (NetCDF) expect files to be accessible on a mounted filesystsem. Our preferred option is to run flows on kubernetes, but it is not clear to us how to mount a volume on the pods that execute the flows. All tips greatly appreciated

Hello! I have the following use case: launch specific container (expose to end user), mount volume and clone gitlab repo there (usage of analytical environment by user) push changes to repo and end running of container. What is the best tool and approach to run such a scenario, with as little as possible user interference? Prefect?

Hi everyone, I was wondering if anyone could kindly give some pointers to a data engineering noob like me. 🤓 I have to ingest about 10GB of data from xlsx files into a sql starschema. The person working on this before me used Pandas to do all the data cleaning, surrogate key creation etc. I am worried that as we get more data that at some point it won't fit into memory anymore and that we will have a problem. 🤪 What are the best approaches for this in your opinion? Do you put the data into some intermediary SQL table? Do you use Dask? (Taking into consideration that I would like to learn/leverage😛refect:).

Hi All! I am trying to use Parameters / Context to control two of my tasks in one Flow. At first to try, I wrote two command lines to get Parameters, e.g.

a = Parameter('a', default=1)

b = Parameter('b', default=2)

In Prefect API, I only see one Parameter for

b

. Did I miss something? If I need to use multiple controls for different arguments in functions, which one do you suggest me to use, Parameter or Context?

Do we need to re-register the flow (using GitHub storage) if we change the default value of a parameter?

Hi - Under what scenarios would a flow take longer to run locally with the DaskExecutor than the LocalExecutor? Is to many small tasks one of them?

<!here> @Dylan I'm trying to reload a settings module in a flow for each scheduled iteration, but it never loads the changes in settings module. Is that because of some restrictions with the flow, that I might be overlooking

with Flow('ETL Status Cron Flow', schedule=CronSchedule("*/5 * * * *")) as cron_flow:
	work_conf = importlib.import_module('settings.work_settings')
	importlib.reload(work_conf)
	my_settings = getattr(work_conf, 'my_settings')                        

	if(my_settings['active'] == False):
		return
	else:
		print('do something')
     
state = cron_flow.run()

This message was deleted.

Hi all, I'm using fargate to set up my task environment and using non-dockerized containers in s3. I can see the task starting on fargate and I'm connected to the cluster, but I do not see the flow running. It's stuck after "Deploying flow run " and "Adding tags for flow_id and version". Not sure why it's stuck 😕 cc: @itay livni

fargate.py

from prefect.agent.fargate import FargateAgent
agent = FargateAgent(
    #
    enable_task_revisions=True,
    launch_type="FARGATE",
    taskRoleArn="arn:aws:iam::#####61:role/ayang-role",
    executionRoleArn="arn:aws:iam::#####61:role/ayang-role",
    family="#####-task",
    cluster="#####-fargate-cluster",
    networkConfiguration={
        "awsvpcConfiguration": {
            "assignPublicIp": "ENABLED",
            "subnets": ["subnet-#####11"],
            "securityGroups": []
            }
    },
    cpu="1024",
    memory="4096",
    containerDefinitions = [
        {
            "logConfiguration": {
                "logDriver": "awslogs",
                "options": {
                    "awslogs-group": "/ecs/my-first-task",
                    "awslogs-region": "us-east-2",
                    "awslogs-stream-prefix": "ecs"
                }
            },
            "image": "#####.<http://dkr.ecr.us-east-2.amazonaws.com/pga_scraper:latest|dkr.ecr.us-east-2.amazonaws.com/pga_scraper:latest>",
            "name": "pga_scraper"
        }
    ],
    labels=["s3-flow-storage"]
)
agent.start()

Hi all, I've got kind of a dumb question. If I already have a bunch of dask.delayed functions, and I'm going to pull them into a prefect, what's the best way to do that? Should I call compute on them within the prefect Task?

Is there any way to separate secrets for different flows? The use case is we have some jobs that interact with our PostgreSQL DB, some are for pulling data so we would like to keep those with read-only credentials and others are ETL so they need read-write. We would like to avoid allowing users to inadvertently create flows that might end up deleting data. Is there a concept of namespace or some other mechanism for segmentation? So for example if Flow A runs and pulls secret

my_password

it'll be different a different password compared to if Flow B pulls the same secret?

Is caching done entirely based off of file existence? Has there been thought into other methods of result caching? I know there it's possible to add a

Serializer

but has there been thought into a

Verifier

or similar? (Entirely just curious at this point)

Hey - I'm having an issue running a Flow within Docker (https://docs.prefect.io/orchestration/tutorial/docker.html#persisting-your-flow-with-docker-storage) I've set

prefect backend server

Started

prefect agent start docker

Copied the Hello Docker flow from the web page above, Registered that flow, it created the image without issue:

`Successfully tagged hello-docker:2020-07-14t05-57-08-042222-00-00
Flow: <http://localhost:8080/flow/e25b220f-c9e4-4051-9f00-20ca88122a99>`

and submit a run via the GUI server running on localhost:8080. The run enters a state of Submitted and does not seem to get any further, it stays in that state.

Hey guys, does anyone have any experience using the FargateTaskEnvironment? Specifically the aws_sesison_token, having problems getting AWs to accept it, I keep getting the following error:

Failed to load and execute Flow's environment: ClientError('An error occurred (UnrecognizedClientException) when calling the RegisterTaskDefinition operation: The security token included in the request is invalid')

Sample config for the environment:

flow.environment = FargateTaskEnvironment(
    launch_type="FARGATE",
    region="us-east-1",
    aws_session_token="FwoGZXIvYXdzEOL//////////wEaDFSgRMmr3M07yXJ3gCKCAdZO6f/LRZc6b7DjDip0lrTvCa+FDQpFAGJEyB6Ka1tF9By3fTgKkbqSM6EnuHQgTEviQJrOn13E7wlvKKVV1++YGaa3gb1Pn9q12BxCN7I6SvQ8oBW9AE73Judo0tuYTdTc5eYC7m2PaYU/d5fkRIj29EJWp9EpO3+yq/S1saxiGvYosKGy+AUyKBOOerH4ymGN9lxo/5aprU5DXumyaC6yg2satDNNoUdPaSBQ2R8fNq0=",
    cpu="256",
    memory="512",
    enable_task_revisions="True",
....
....
)

Hi The tutorials, both the "Running a Flow" and the "Universal Deploy" tutorials, fail. Both on the 

prefect auth login

 step. Prefect tries to connect locally rather than to your cloud:

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f5aa8c9f630>: Failed to establish a new connection: [Errno 111] Connection refused',))

How do I tell Prefect that it should connect to the cloud?

@Philip MacMenamin Came here to report exactly the same issue. The task state doesn't change. It seems to work quite alright on the cloud.

Hey guys, I have a question regarding triggering

CloudFlowRunner

flows. Locally I have a flow setup and registered to the cloud backend. I can trigger the flow via the web-ui, however when running locally I get the following:

>> cloud_flow = CloudFlowRunner(flow)
>> flow_state = cloud_flow.run(return_tasks=flow.tasks)
>> flow_state
[2020-07-14 09:37:58] INFO - prefect.CloudFlowRunner | Beginning Flow run for 'Map Reduce'
<Failed: "Could not retrieve state from Prefect Cloud">

Am I missing something here?

Hey folks, We are still new to prefect and have a question in terms of code organization: Do you have an extra repository for prefect flows (⬅️) or do you save the prefect flows within other repositories (➡️)? For ⬅️ (extra repo), how do you import the Python packages? And in general, how does your CI/CD look like? We are setting up a backend from scratch (using GitLab, pulumi (or terraform), dbt, snowflake and sigma (or maybe looker, but might be too expensive..), so we are happy to hear best practices and learn from you! 🙂

Hey all - I am using Secret like this:

task = ShellTask()

m = Secret('my_secret_on_the_cloud').get()

with Flow('my_flow') as flow:
    name = task(
        helper_script='cd ~'
        env={'CREDENTIAL':m},
        command="mysqldump -u my_user -p$CREDENTIAL my_db > my_db.sql"
        )

while I should be using it like this:

task = ShellTask()

with Flow('my_flow') as flow:
    m = Secret('my_secret_on_the_cloud')
    name = task(
        helper_script='cd ~'
        env={'CREDENTIAL':m},
        command="mysqldump -u my_user -p$CREDENTIAL my_db > my_db.sql"
        )

The later fails with

TypeError

. I have the secret on the cloud and in the local

config.toml

for testing purposes. What am I doing wrong?

Hello,   I am new to prefect and am trying to use DbtShellTask which does work. The issue I am having is that if the dbt fails the error is not the full message compared to if I were to run it manually. For example if compiling of the dbt fails it will return compilation error but excludes the file that the error occurs in. Running manually gives the same error message and include which file the error is in.   Another example is if the dbt fails during execution running manually I get the following messages:  

10:33:16 | Concurrency: 1 threads (target='dev')
10:33:16 |
10:33:16 | 1 of 4 START table model dbt_test.add_rid............................ [RUN]
10:33:28 | 1 of 4 OK created table model dbt_test.add_rid....................... [SELECT 161069 in 11.75s]
10:33:28 | 2 of 4 START table model dbt_test.union.............................. [RUN]
10:33:36 | 2 of 4 OK created table model dbt_test.union......................... [SELECT 159280 in 7.86s]
10:33:36 | 3 of 4 START table model dbt_test.union_join_test.................... [RUN]
* Deprecation Warning: The adapter function `adapter.get_columns_in_table` is
deprecated and will be removed in a future release of dbt. Please use
`adapter.get_columns_in_relation` instead.
Documentation for get_columns_in_relation can be found here:
   <https://docs.getdbt.com/docs/adapter>
10:33:36 | 3 of 4 ERROR creating table model dbt_test.union_join_test........... [ERROR in 0.22s]
10:33:36 | 4 of 4 START table model dbt_test.union_join......................... [RUN]
10:33:50 | 4 of 4 OK created table model dbt_test.union_join.................... [SELECT 159280 in 14.28s]
10:33:50 |
10:33:50 | Finished running 4 table models in 35.12s.

  When I run from DbtShellTask for the same dbt workflow I get this message:  

July 14th 2020 at 7:37:14am | prefect.DbtShellTask
ERROR lens
Command failed with exit code 1: Done. PASS=3 WARN=0 ERROR=1 SKIP=0 TOTAL=4

  Trying to make it easier to track down the error. Is there anyway to include these full error messages?   Here is a sample of how I am doing this. I was hoping return_all would return all messages from dbt.  

with Flow(name="dbt_flow") as flow:
   task = DbtShellTask(
       profile_name='default',
       environment='dev',
       dbt_kwargs={
               'type': 'postgres',
               'threads': 1,
               'host': 'IP',
               'port': 5433,
               'user': 'username',
               'pass': 'docker',
               'dbname': 'actualDbName',
               'schema': 'dbt_test'
       },
       overwrite_profiles=True,
       profiles_dir='Actual Path', return_all=True
   )(command='dbt run' )
 
flow.register()

  Any help would be great.   P

Hey guys, Is it possible to trigger one flow to run another? If so, is there a prefect recommended way of doing it? I have a workflow where there are some heavyweight tasks that do not change much with time (related to parsing and data normalization). Tasks downstream from these are driven by research and regularly require re-running. As Flow seems to be atomic when it comes to scheduling in prefect, what is the best way for me to structure my workflow in prefect? Thanks in advance.

Is there a good way to handle situations where a Prefect agent may shut down during flow execution? I'm looking into using EC2 spot instances as Prefect agents. These instances can theoretically be shut down without warning. Ideally, any Flows that were running would be aware that something has disrupted execution, and signal that the flow needs to be re-run. If I try killing local agent process (Ctrl + C while the agent is mid-flow execution), it seems like tasks hang in the

Submitted

state by default.

This is a pathological case, but it seems possible to end up with multiple tasks in the same flow having the same slug:

import prefect as p

@p.task
def add_one(n):
    return n + 1

@p.task
def add_two(n):
    return n + 2

with p.Flow('f') as f:
    a = p.Parameter('a')
    b = add_one(a)
    b.slug = 'my_slug'
    c = add_two(a)
    c.slug = 'my_slug'

f.get_tasks(slug='my_slug') # returns both tasks

f.validate(). # passes

Hello everyone. I am trying to access a Parameter from within a state handler using prefect.context.get("parameters", {}).get(stakeholderEmail). When I run my code I get no errors, but it does result in my variable containing the emails to be a nonetype. Is my approach the proper method for retrieving a parameter from inside a state handler, or is there a better way to do this?

Is there any documentation about how you would go about setting up prefect server to run in a containerized env? I saw Tyler Wanner's terraform setting up a GCP compute instance, but is there a way to set things up to get run out of a container?

Hi All, I am evaluating Prefect Core and I love the look of it at the moment and have some questions around deployments, flow registration and versioning. We currently have a Build/Deploy process that is using Azure Pipelines and I anticipate that given a repo it would bundle up the relevant flows and python files. For deployment I would like to deploy to AWS S3 buckets with a version number. I do see that there is a storage option but that would require the deploy agent to have prefect/python installed to call flow.build with the s3 storage option set. In terms of hosting I was hoping to self host in ECS the prefect server/UI and relevant agents 1. Is there an option where I can deploy versioned files myself using either terraform or aws cli to S3, if so how do I register the relevant flows? 2. I don't want my deployment pipeline to have a VPN to the relevant AWS VPC/environment so can flow.register() be completely separated from flow.build()/etc 3. For flows deployed to S3, how does the server ensure the latest version is being used/registered? 4. My AWS environments are in completely separate accounts/VPCs, and I noticed Prefect environments/agents/flows have labels, is there a particular pattern that ensures flows are reusable between environments?

I'm having issues deleting a secret. I added one with a /, e.g

my_project/postgres_user

that got converted to

my_project/

and now I cannot delete it.

Hi folks, Have been checking the page about dask cluster in K8s (https://docs.prefect.io/orchestration/recipes/k8s_dask.html#dask-cluster-on-kubernetes), and there is an env variable that I cannot figure out why it is useful. Nor finding any info about it. Can someone explain what it is used for :

DASK_DISTRIBUTED__SCHEDULER__BLOCKED_HANDLERS

?