from time import sleep

from prefect import Flow, apply_map, task
from prefect.executors import DaskExecutor


@task()
def test_1(x):
    sleep(2)
    print(f"test_1 with {x}")
    return x


@task()
def test_2(x):
    sleep(2)
    print(f"test_2 with {x}")


def micro_flow(x):
    test_1_task = test_1(x)
    test_2(test_1_task)


with Flow(
    "example",
    executor=DaskExecutor(cluster_kwargs={"n_workers": 1, "threads_per_worker": 2}),
) as flow:
    apply_map(micro_flow, range(10))


if __name__ == "__main__":
    flow.run()
    # flow.visualize()

KubernetesPodOperator

Sensors

Need help here. Trying to do in python as below. This is to check the flow state and cancel in the flow is in running state. How to pass the values to "$flowRunId: UUID!"                                                                                                                        import prefect
from prefect import Client
from prefect import task, Flow
@task()
def check_runs():
  c = Client()
  query = """
  query RunningFlowsName {
  flow(where: {name: {_eq: "flowstatechecktest"}}) {
    id
  } }  """
  print('======')
  print(c.graphql(query=query))

  query2 = """
  query RunningFlowsState {
  flow_run(where: {state: {_eq: "Running"}}) {
    state
  }  }  """
  print('======')
  print(c.graphql(query=query2))

  query3 = """
  mutation CancelFlowRun($flowRunId: UUID!) {
  cancel_flow_run(input: {flow_run_id: $flowRunId}) {
    state
  }  }  """
  c.graphql(query=query3)

with Flow("flowstatechecktest") as flow:
    check_runs()
flow.run()

@task(max_retries=3, retry_delay=timedelta(minutes=30))

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

Heartbeat process died with exit code -9

Failed to set task state with error: ClientError([{'message': 'State update failed for task run ID 43f52f19-fffb-4d16-8223-da4ffc5668b2: provided a running state but associated flow run 8c8fc810-eb3d-447c-ab70-76dd1dc2acaa is not in a running state.', 'locations': [{'line': 2, 'column': 5}], 'path': ['set_task_run_states'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'State update failed for task run ID 43f52f19-fffb-4d16-8223-da4ffc5668b2: provided a running state but associated flow run 8c8fc810-eb3d-447c-ab70-76dd1dc2acaa is not in a running state.'}}}],)

HdfsDataFrameResult

PREFECT__FLOWS__CHECKPOINTING=true

read

write

exists

run

DataFrame

Flow

TaskA(checkpoint=True, result=HdfsDataFrameResult(...))

HdfsDataFrameSerializer

from prefect.tasks.azure.blobstorage import (
    BlobStorageDownload
)

storage = BlobStorageDownload(azure_credentials_secret="AZURE_DEMO", 
                              container="mycontainer")

storage.run(blob_name="demo")

ValueError: Local Secret "AZURE_DEMO" was not found.

Results

PostgresExecuteMany

from prefect import task, Flow
import prefect.tasks.postgres.postgres as pg

DB_NAME = 'blah'
DB_USER = 'blah'
DB_HOST = 'localhost'

sql_insert = pg.PostgresExecuteMany(db_name=DB_NAME, user=DB_USER, host=DB_HOST)

@task()
def extract_data():
    vals = ['aaa', 'bbb']
    return vals


@task()
def load(data):
    insert_stmt = """INSERT INTO "table_a" ("COL_A", "COL_B") VALUES (%s, %s)"""
    ret = sql_insert(query=insert_stmt, data=data, commit=True)
    return ret


def build_flow():
    with Flow('Test ETL') as f:
        data = extract_data()
        load(data)
    return f


flow = build_flow()
flow.run()

[2021-02-24 12:14:55-0800] WARNING - prefect.TaskRunner | Task 'MyTask': Can't use cache because it is now invalid

Result

my_task: MyTask = MyTask(
    checkpoint=True,
    cache_for=dt.timedelta(hours=12),
    result=HdfsDataFrameResult(spark=spark, location=hdfs_path_for_this_task),
)

my_task(…)

cache_key

S3Results

[2021-02-24 14:11:49-0800] ERROR - prefect.TaskRunner | Task 'plot_map_counts_data[0]': Unexpected error while running task: TypeError("exists() got multiple values for argument 'location'",)

@task(
    checkpoint=True,
    target="{flow_name}/{today}/{task_name}_{map_index}.png",
    result=prefect.engine.results.S3Result(bucket=results_s3_bucket),
)
def plot_map_counts_data(df, location, resource):
    #some task
    return image

with Flow("example") as flow:

    plot_map_counts_data.map(
        df,
        locations,
        unmapped("test_resource"),
    )

@task
def get_url(run_id, hook):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("RUN ID: %s", run_id)
    url = hook.get_run_page_url(run_id=run_id)
    return url

SubmitRun = DatabricksSubmitRun()

with Flow("test_databricks", storage=STORAGE, run_config=RUN_CONFIG) as flow:
    conn = PrefectSecret('DATABRICKS_CONNECTION_STRING')
    json = get_job_config()
    run_id = SubmitRun(json=json, databricks_conn_secret=conn)
    hook = SubmitRun.get_hook()
    url = get_url(run_id, hook)

ERROR:prefect.TaskRunner:Unexpected error: TypeError("argument of type 'NoneType' is not iterable")
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 863, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/prefect/utilities/executors.py", line 298, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/rbastian/enverus/RAI/prefect/flows/test-databricks.py", line 32, in get_url
    url = hook.get_run_page_url(run_id=run_id)
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/prefect/tasks/databricks/databricks_hook.py", line 248, in get_run_page_url
    response = self._do_api_call(GET_RUN_ENDPOINT, json)
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/prefect/tasks/databricks/databricks_hook.py", line 148, in _do_api_call
    if "token" in self.databricks_conn:

{
  "kind": "Status",
  "apiVersion": "v1",
  "metadata": {},
  "status": "Failure",
  "message": "Operation cannot be fulfilled on resourcequotas \"gke-resource-quotas\": the object has been modified; please apply your changes to the latest version and try again",
  "reason": "Conflict",
  "details": {
    "name": "gke-resource-quotas",
    "kind": "resourcequotas"
  },
  "code": 409
}