Hi
i'm currently testing out deploying metaflow to...
# ask-metaflowr
refined-zebra-23040
09/12/2024, 10:13 AMHi
i'm currently testing out deploying metaflow to airflow, but my converted airflow dags are never actually running.
Locally the workflow does work.
Any help would be very appreciated.
Thanks
✅ 1
a
ancient-application-36103
09/12/2024, 2:50 PM@refined-zebra-23040 what is the error that you see?
r
refined-zebra-23040
09/12/2024, 3:27 PM[2024-09-12T15:25:43.232+0000] {kubernetes_executor_utils.py:157} INFO - Event: and now my watch begins starting at resource_version: 0a
ancient-application-36103
09/12/2024, 3:28 PMDoes this happen for every flow or just flow? Can you try a very simple linear flow? Also, can you help us with what your airflow setup looks like and how is it connected to your Kubernetes cluster?
r
refined-zebra-23040
09/12/2024, 3:31 PMsimpler flows e.g. parameter flow is working fine, funny thing is that it instantly is in state "success" without actually running any tasks.
a
ancient-application-36103
09/12/2024, 3:32 PMThe simpler flows have the same behavior? ^
r
refined-zebra-23040
09/12/2024, 3:34 PMno the simpler flows are working fine (linear flow, parameter flow etc.)
a
ancient-application-36103
09/12/2024, 3:35 PMGotcha - is there an example flow you are able to share with us that fails?
r
refined-zebra-23040
09/12/2024, 3:41 PM Copy code
from metaflow import FlowSpec, step, environment, kubernetes, conda
import os
class ClickhouseQueryFlow(FlowSpec):
@environment(vars={
"CLICKHOUSE_HOST": os.getenv("CLICKHOUSE_HOST"),
"CLICKHOUSE_USER": os.getenv("CLICKHOUSE_USER"),
"CLICKHOUSE_PASSWORD": os.getenv("CLICKHOUSE_PASSWORD"),
"CLICKHOUSE_PORT": os.getenv("CLICKHOUSE_PORT"),
})
# @kubernetes
# @pypi(packages={'clickhouse-connect': '0.7.19'})
@conda(packages={'clickhouse-connect': '0.7.19'})
@step
def start(self):
import clickhouse_connect
"""
This step connects to ClickHouse and executes a query.
"""
self.host = os.getenv('CLICKHOUSE_HOST')
self.user = os.getenv('CLICKHOUSE_USER')
self.password = os.getenv('CLICKHOUSE_PASSWORD')
self.port = os.getenv('CLICKHOUSE_PORT')
# Initialize ClickHouse connection
client = clickhouse_connect.get_client(
host=self.host,
username=self.user,
password=self.password,
port=self.port,
secure=True
)
# Example query: fetching some data
query = "SELECT * FROM system.tables;"
self.results = client.query(query).result_rows
print("Query executed successfully!")
print(self.results)
self.next(self.process_results)
# @kubernetes
@step
def process_results(self):
"""
Process the results fetched from ClickHouse.
"""
# For this example, we'll just print the results.
print("Processing results...")
for row in self.results:
print(row)
self.next(self.end)
@step
def end(self):
print("Flow completed!")
if __name__ == '__main__':
ClickhouseQueryFlow()refined-zebra-23040
09/12/2024, 3:41 PMyes here, thanks for the help
a
ancient-application-36103
09/12/2024, 3:43 PMDoes this flow execute if you don’t have environment decorator?
r
refined-zebra-23040
09/12/2024, 4:36 PMi updated the simple flow and now no flow is running, i guess this has something to do with my env. variables, as this is the only thing that changed in between.
3 Views