prefect-community #prefect-community

Emerson Franks

07/12/2022, 5:51 PM

Hey there Prefect folks! I'm getting a Prefect -> FiveTran POC up and encountering an issue where the flow runs fine locally from terminal but doesn't appear in the UI, and therefore can't be run in the UI, once I launch prefect agent local start. I'm using 1.0 as I didn't see a FiveTran module for 2.0. If anyone can point me at what I'm missing I would greatly appreciate it!

✅ 1

Kevin Grismore

07/12/2022, 5:58 PM

I'm trying to use the Scrapy framework in conjunction with Prefect 2.0, and Scrapy automatically sets up its own core Python Loggers. When I run a flow in the terminal, I can see the logs coming from Scrapy's loggers like so:

13:37:49.758 | DEBUG   | scrapy.core.scraper - Scraped from <200 <https://quotes.toscrape.com/page/9/>>

However, those logs don't appear in the flow run logs UI. I'm guessing I need to set

PREFECT_LOGGING_EXTRA_LOGGERS

, but I'm not certain what the right way to do that is.

prefect config set PREFECT_LOGGING_EXTRA_LOGGERS="scrapy"

seems to have no effect. I also tried

prefect config set PREFECT_LOGGING_EXTRA_LOGGERS="scrapy.core.engine"

using the name of the specific logger I wanted to see, but that didn't work either.

✅ 1

Jacob Bedard

07/12/2022, 5:59 PM

Anybody ever seen this before? I'm running a local agent in a conda env with python 3.8.13, and it's the same one I registered the flow on. The unpickling seems to be the issue, but I'm really confused about the other python version mentioned in this error msg.

Copy code

Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'snowflake\'")\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n  - python: (flow built with \'3.8.13\', currently running with \'3.8.10\')\nThis also may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

I created another flow that logged the environment, then ran that one via the UI too. It reported that it was running in the right env. Where is this 3.8.10 version coming from? And what's the solution for the issue of not being able to pull up snowflake? I didn't have this issue with my last prefect setup, so I'm just kinda scratching my head here

✅ 1

Josh

07/12/2022, 6:11 PM

Anyone ever run into this error before?

Unexpected error: TypeError("InvalidSchema.__init__() missing 2 required positional arguments: 'local_schema' and 'remote_schema'")

I’m passing dataframes down to another task and for some reason it’s not able to startup the task at all.

✅ 1

Constantino Schillebeeckx

07/12/2022, 6:16 PM

my Prefect cloud tenant is in a strange state where I cannot create a project because it already exists, but i cannot delete it because it doesn't exist

✅ 1

Andrew Pruchinski

07/12/2022, 6:30 PM

Hey all! Thanks in advance for your time. Looking how we can use the python prefect client to access the project name the current flow is being ran under. This is how we will determine which environment our processes are being ran under. We used to have different Tenants for the environments and used the

TenantView

class but for security purposes we now have all environments (projects) under one tenant/team. Appreciate the help!

✅ 1

Irisa Shrestha

07/12/2022, 7:03 PM

Hi guys! I am new to Prefect cloud and I need your help! I have just created a project that extracts data from a website and updates my database, I need to schedule this to happen every week. Currently, I am using a local agent, however, I was wondering what would be a better agent for me so that I don't have to run this agent locally every time I need my project flow to run? Thank you so much!

✅ 1

John Mil

07/12/2022, 8:27 PM

Hello, where do i post if i am encountering issues with graphql?

✅ 1

Kevin Grismore

07/12/2022, 9:26 PM

how might I make use of a serializer to package the module my

flow.py

is contained in?

✅ 1

davzucky

07/13/2022, 12:07 AM

Question about Precfect 2. Reading more and more about the new concept of Block. I like the idea and what it provide you. This will be powerful. • Will block replace the concept of filesystem? • How can I know the capability of a block? • Can I use the block concept to hide the storage location depending on the setup I do (local dev vs prod for example)?

✅ 1

wonsun

07/13/2022, 1:55 AM

Hi all! Is this correct to use prefect's secret function? By following the prefect's docs, but i got the message like this.

Copy code

Traceback (most recent call last):
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 551, in _send_request
    response.raise_for_status()
  File "C:\Users\user\anaconda3\lib\site-packages\requests\models.py", line 960, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://localhost:4200/>

This is the part of my code.

Copy code

from prefect import Flow, task, resource_manager, case
from prefect.tasks.control_flow import merge
from prefect import Client
from prefect.tasks.secrets import PrefectSecret

client = Client()
client.set_secret(name="secret", value="<mysql+pymysql://wonsun>:RKSek849200!@192.168.0.13/metadata")

wonsun

07/13/2022, 2:16 AM

Hi all! Is this correct to use prefect's secret function? By following the prefect's docs, but i got the message like this. (full of error message)

Copy code

Traceback (most recent call last):
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 551, in _send_request
    response.raise_for_status()
  File "C:\Users\user\anaconda3\lib\site-packages\requests\models.py", line 960, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://localhost:4200/> 

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File ".\label_test.py", line 19, in <module>
    client.set_secret(name="mysql", value="<mysql+pymysql://user:pwd@host/database>")
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 1643, in set_secret
    result = self.graphql(
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 452, in graphql
    result = <http://self.post|self.post>(
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 407, in post
    response = self._request(
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 641, in _request
    response = self._send_request(
  File "C:\Users\user\anaconda3\lib\site-packages\prefect\client\client.py", line 564, in _send_request
    raise ClientError(f"{exc}\n{graphql_msg}") from exc
prefect.exceptions.ClientError: 400 Client Error: Bad Request for url: <http://localhost:4200/>

The following error messages were provided by the GraphQL server:

    GRAPHQL_VALIDATION_FAILED: Unknown type "set_secret_input". Did you mean
        "agent_set_input", "edge_set_input", "log_set_input", "task_set_input", or
        "agent_insert_input"?
    GRAPHQL_VALIDATION_FAILED: Cannot query field "set_secret" on type "Mutation".

The GraphQL query was:

    mutation($input: set_secret_input!) {
            set_secret(input: $input) {
                success
        }
    }

The passed variables were:

    {"input": {"name": "mysql", "value": "<mysql+pymysql://user:pwd@host/database>"}}

~~And here is my code of the part that I think is a problem.~~ (code block moved to thread) The information to be contained in 'secret' is connection information that can access the database used. I used the Prefect 1.2.2. In this common case, How can i use the secret fuction well?

✅ 1

Rajeshwar Agrawal

07/13/2022, 10:05 AM

Hey Prefect! I executed a KubernetesRun flow using

config.toml

placed at

/home/.prefect/config.toml

Is there a way to check via prefect-ui or apollo whether the config items from this toml are applied?

✅ 1

Tom Klein

07/13/2022, 10:38 AM

Hey 🙋 is it somehow possible to do something in between a map and a reduce? 🤔 e.g. - let’s say i have the following pattern: • a huge input CSV (let’s say 200K rows) • that i then split into a ton of tiny CSVs (let’s say 1000 rows each) • then some work is done on each one, and a new output CSV is generated per each tiny CSV • and afterwards i wanna combine the outputs - but in “slightly bigger batches”, i.e. - let’s say in groups of 5 or groups of 10 - and each one of those to turn into a “medium sized” CSV (with 5K or 10K rows) which is then uploaded to S3 obviously, i don’t want the last step (4) that combines them to wait for ALL the tiny CSVs to be processed (so i don’t want it to just be a simple reduce), just enough (e.g. 5, or 10) that could be grouped and uploaded does that make any sense?

✅ 1

Enrique

07/13/2022, 10:39 AM

Hi all! 🙂 I have a question (maybe it's stupid but here it goes xD) I'm working with Prefect in a Kubernetes environment, which means that the agent and the jobs are running in the cluster. The cluster has 2 CPUs. When I'm trying to run three parallel tasks, only two of them run and then when one of them ends the third one starts running. Is that the expected behavior or I'm doing something wrong? Thanks a lot for your tool, Prefect rocks! 😎

😎 1

✅ 1

Jason Thomas

07/13/2022, 12:11 PM

Hi all, version 2.0b8 I’m playing with flow retries on and got a result I didn’t expect. I’ll post code in the thread, but here’s what’s happening: - flow has one retry - I call flow, passing in a list of 4 unprocessed ‘files’ - retry 0: - flow filters the list to remove any processed files, leaving 4 files to process - flow iterates over the filtered list, calling task on each unprocessed file - task processes each file - on one file, task raises an error - flow continues to run, processing 3 out of 4 files - flow finishes in state

AwaitingRetry

- retry 1: - flow filters the list, leaving 1 file to process - flow iterates over the filtered list, but does not call task - flow finishes in state
Completed
I’m guessing task is not called on the retry because the result has been cached and there is no reason to try again. That’s fine. However, I expected the final state of the task to be
Failed
. Is this behaving as intended?

✅ 1

Tom Klein

07/13/2022, 12:40 PM

Hi again 😄 I just got the error

No heartbeat detected from the remote task; marking the run as failed

- this happend for tasks being run in parallel via the dask

KubeCluster

looking at our (internal) logs - it seems like all of them died simultaneously, 20+ minutes into their run there’s a total of 58 tasks being run, with 4 workers the first 4 tasks took approx. 1 hour to run and completed successfully, and then the next 4 that were run all seem to have failed simultaneously any idea / help?

✅ 1

Shivam Bhatia

07/13/2022, 2:12 PM

Hi, I am trying to figure out running flows on a docker container (prefect 2.0b8) After the flow run is complete I get an error and my agent stops running. (Error log in the thread) How can i fix this? Any help would be appreciated

✅ 1

ibrahem

07/13/2022, 2:20 PM

Hi ... I'm having an issue with the local agent it has been working fine for days and suddenly after restarting the prefect server the agent wont run!. when I do prefect local agent start it seems that it runs okay in terminal (agent registration successful) but after a while I get this error: requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by ReadTimeoutError("HTTPConnectionPool(host='localhost', port=4200): Read timed out. (read timeout=15)")) Appreciate your help

✅ 2

Kevin Grismore

07/13/2022, 3:35 PM

I'm trying to create a deployment for Prefect Cloud 2.0 for a flow that imports a local package named

tutorial

that's in the flow's directory, and when I run

prefect deployment create deployment.py

I get

ModuleNotFoundError: No module named 'tutorial'

✅ 1

Josh Paulin

07/13/2022, 3:55 PM

If I have a conditional check in my flow thats something like

Copy code

with case(cond, True):
    val1_if_true = action1_if_true()
    val2_if_true = action2_if_true()


with case(cond, False):
    val_if_false = action_if_false()

What (if anything) do I pass to the merge function?

✅ 1

Sabir Ali

07/13/2022, 5:21 PM

I am trying to connect elasticsearch clietn but getting exception

Copy code

Traceback (most recent call last):
  File "/Users/sabirali/PycharmProjects/ETL/ElasticSearchClientTest.py", line 5, in <module>
    print(<http://client.info|client.info>())
  File "/Users/sabirali/PycharmProjects/ETL/venv/lib/python3.8/site-packages/elasticsearch/_sync/client/utils.py", line 414, in wrapped
    return api(*args, **kwargs)
  File "/Users/sabirali/PycharmProjects/ETL/venv/lib/python3.8/site-packages/elasticsearch/_sync/client/__init__.py", line 2277, in info
    return self.perform_request(  # type: ignore[return-value]
  File "/Users/sabirali/PycharmProjects/ETL/venv/lib/python3.8/site-packages/elasticsearch/_sync/client/_base.py", line 332, in perform_request
    raise UnsupportedProductError(
elasticsearch.UnsupportedProductError: The client noticed that the server is not Elasticsearch and we do not support this unknown product

I used following command to install elasticsearch client

Copy code

(venv) sabirali@Sabirs-MacBook-Pro ETL % pip install elasticsearch      
Collecting elasticsearch
  Using cached elasticsearch-8.3.1-py3-none-any.whl (382 kB)
Requirement already satisfied: elastic-transport<9,>=8 in ./venv/lib/python3.8/site-packages (from elasticsearch) (8.1.2)
Requirement already satisfied: urllib3<2,>=1.26.2 in ./venv/lib/python3.8/site-packages (from elastic-transport<9,>=8->elasticsearch) (1.26.9)
Requirement already satisfied: certifi in ./venv/lib/python3.8/site-packages (from elastic-transport<9,>=8->elasticsearch) (2022.6.15)
Installing collected packages: elasticsearch
Successfully installed elasticsearch-8.3.1
WARNING: You are using pip version 21.3.1; however, version 22.1.2 is available.
You should consider upgrading via the '/Users/sabirali/PycharmProjects/ETL/venv/bin/python -m pip install --upgrade pip' command.
(venv) sabirali@Sabirs-MacBook-Pro ETL %

✅ 1

Divya

07/13/2022, 6:33 PM

Hello all, I am trying to connect to Perfect Server but I am getting the error message as below. Can you please help me figure out the issue or the step that I am missing? The steps taken so far: 1. Installed prefect using the pip command 2. Created a flow and ran successfully(screenshot attached). 3. Executed the below commands to start the server: *prefect backend serve*r prefect server start --detach --use-volume --expose Thank you! Best, Divya

✅ 1

Frank Hardisty

07/13/2022, 8:25 PM

Hello, I work at Replica, we are enthusiastic Prefect users on a paid plan. Yesterday we had an unfortunate misconfiguration which caused us to send an excessive amount of logs to Prefect, and our Prefect logging was shut off. Fair enough. The misconfiguration was fixed, we are not spamming logs anymore. however the lack of Prefect logs is an operational issue for us, including for some Flows that produce customer-facing data. So... can I ask for our logging to be turned on? Or an estimate of when it might be turned back on? Thanks in advance. If this is not the right place to ask this question, apologies, please direct me to the right place to ask about this. Thanks.

✅ 1

Victoria Alvarez

07/13/2022, 8:47 PM

Hello. I run a flow every day using bigquery_insert_stream. I am having some trouble running this today because I get timeout or 500 internal server error. Any idea of what's wrong? Thank you in advance!

✅ 1

Ryan

07/14/2022, 2:05 AM

Hi all -- are there plans to add a 'Paused' state / signal / other mechanism to 2.0 for flows and/or tasks?

✅ 1

Amanda Wee

07/14/2022, 2:46 AM

Is Prefect angling for an in-road to join Airflow as a managed solution by AWS? I'm asking because I've recently joined a new company, and the current plan is to use AWS managed Airflow to orchestrate dbt. I'm still learning, but my assessment is that it doesn't make sense to propose Prefect as an alternative right now because what the team is doing with Airflow is so simple. But in the future, there may be more, in which case I expect to run into the lack of AWS managed Prefect as an obstacle.

✅ 1

Максим Пышный

07/14/2022, 2:50 AM

Hi, folks! Is there any way to get failed task's state message (or failed task's name at least) inside flow-level state handler? I am trying to build a message for slack alert that contains information about particular reason of fail. But I want to try to declare state handler just once (e.g. on flow level).

✅ 1

Shivam Bhatia

07/14/2022, 7:46 AM

Hi, I just hosted a prefect 1.2.4 on a gcp vm I am trying to run a flow using VertexRun and I got this error

Copy code

400 List of found errors:	1.Field: job_spec.worker_pool_specs[0].container_spec.env[12].value; Message: Required field is not set.	2.Field: job_spec.worker_pool_specs[0].container_spec.env[3].value; Message: Required field is not set.	3.Field: job_spec.worker_pool_specs[0].container_spec.env[4].value; Message: Required field is not set.	 [field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[12].value"
  description: "Required field is not set."
}
field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[3].value"
  description: "Required field is not set."
}
field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[4].value"
  description: "Required field is not set."
}
]

Do i need to configure vertex jobs manually?

✅ 1

Faheem Khan

07/14/2022, 8:54 AM

Hi All! Prefect2.0b7 Dockerflowrunner, dasktaskrunner. I am getting the following error

_pickle.PicklingError: Can't pickle <function db_con at 0x7f85c986f880>: it's not the same object as __main__.db_con

when I run a task with querying a database. the code runs fine without that db connection task.

✅ 1