flow_1 (preprocessing) -> flow_2 (training) -> flow_3 (deployment)

Hello, I want to schedule a flow all 7 days a week, but on weekend want to reduce the run frequency to every 4 hours once instead of every one hour. This is the pice of code for all 7 days week to run every 1 hr from 6 am to 6 pm which needs modification for weekend schedule.                                                                                                                                                schedule = Schedule(
    # emit an event every hour
    clocks=[
        IntervalClock(
        start_date=pendulum.datetime(2021, 4, 1, tz="America/Los_Angeles"),
        interval=timedelta(hours=1))         
        ],
    
    # only include weekdays
    #filters=[filters.is_weekday],

    # only include 6am and 6pm
    filters=[
        filters.between_times(time(6), time(18))
    #    filters.between_times(time(6), time(18))
    ]
)

Failed to load and execute Flow's environment: StorageError('An error occurred while unpickling the flow:\n AttributeError("type object \'Members\' has no attribute \'ids\'")\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n - prefect: (flow built with \'0.14.15\', currently running with \'0.14.12\')')

prefect register

prefect build

prefect build

prefect register --json

prefect register -p flow.py

flow.register(…)

python flow.py

__main__

flow.run()

extract

At least one upstream state has an unmappable result

@task
def transform(x):
    return x + 30



@task
def extract():
   return [200, 400, 500]



with Flow("flow-name", storage=S3(bucket="bucket_name")) as flow:
    e = extract()
    t = transform.map(e) # fails when i trigger flow in the UI

flow_database = StartFlowRun(flow_name="database", project_name="test", wait=True)
flow_transform = StartFlowRun(flow_name="transform", project_name="test", wait=True)

with Flow("parent-flow") as flow:
   
    data = flow_database() #get a dataframe from flow

    flow_transform(data) # take dataframe and use it another flow

enviornment=env_vars

flow.run_config = DockerRun(env={f"PREFECT__CONTEXT__SECRETS__{k}": v for k, v in config.context.secrets.items()})

set_flow_group_schedule

flow.register()

prefect register

S3

ECS

Docker

docker run

from prefect.schedules import IntervalSchedule, filters

ist = pendulum.timezone('Asia/Calcutta')
schedule = Schedule(
    clocks=[IntervalClock(interval=datetime.timedelta(minutes=1))],
    filters = filters.between_times(pendulum.time(6),pendulum.time(10))

build=True

if

True

, the flow's environment is built prior to serialization; defaults to

True

UserWarning: A flow with the same name is already contained in storage; if you changed your Flow since the last build, you might experience unexpected issues and should re-create your storage object.
  serialized_flow = flow.serialize(build=build)

build=False

build=True

build=False

There was a problem creating your schedule, please try again shortly. Error message: Error: GraphQL error: Invalid clock provided for schedule: {'type': 'CronClock', 'cron': '0 * * * *', 'parameter_defaults': {...}, 'start_date': {'dt': '2021-04-14T14:21:57.622602', 'tz': 'Europe/Amsterdam'}}

@task
def donothing(x):
    pass

with Flow('useless') as flow:
    lst = list(range(4000))
    donothing.map(lst)


flow.executor = DaskExecutor('<tcp://localhost:8786>')

dask-worker --nthreads=50

run_command

RunNamespacedJob

with Flow as flow:
    run_command.map([cmd1, cmd2,...])