#!/usr/bin/env python
# coding: utf-8

import pyodbc
from prefect.storage import GCS
import json
import time
import os
import datetime
from google.cloud import pubsub_v1
import pymysql
from prefect import task, Flow, Parameter



@task(log_stdout=True)
def extract_metadata(source_system,source_database_id):
    
    #provides a list of metadata to loop through and extract delta
    return metadata


def delta_push(metadata):
    # This function extract data from every metadata detail provided and pushes it further

    


@task
def delta_push_wrapper(metadata):
#Looping through ever metadata row and calling the push function 
    for metadata_row in metadata:
        delta_push(metadata_row)



with Flow("data_flow") as flow:
    flow.storage = GCS(bucket="bucketname")
    parameter1 = Parameter("paramater1",default="default")
    parameter2 = Parameter("parameter2",default=1)
    metadata = extract_metadata(parameter1,parameter2)
    delta_push_wrapper(metadata)
    

flow.register(project_name="test_project")

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