Scheduling Flows

from metaflow import FlowSpec, schedule, step

@schedule(daily=True)
class DailyReportFlow(FlowSpec):
    
    @step
    def start(self):
        print("Running daily report...")
        self.next(self.end)
    
    @step
    def end(self):
        print("Report complete")

if __name__ == '__main__':
    DailyReportFlow()

python daily_report.py step-functions create

# Hourly execution
@schedule(hourly=True)
class HourlyFlow(FlowSpec):
    pass

# Weekly execution
@schedule(weekly=True)
class WeeklyFlow(FlowSpec):
    pass

# Run every weekday at 9 AM UTC
@schedule(cron='0 9 ? * MON-FRI *')
class WeekdayMorningFlow(FlowSpec):
    pass

# Run on the first day of every month
@schedule(cron='0 0 1 * ? *')
class MonthlyFlow(FlowSpec):
    pass

# Run every 15 minutes
@schedule(cron='0/15 * * * ? *')
class FrequentFlow(FlowSpec):
    pass

cron(Minutes Hours Day-of-month Month Day-of-week Year)

# Every day at 2:30 PM UTC
@schedule(cron='30 14 * * ? *')

# Every Monday at 8 AM UTC
@schedule(cron='0 8 ? * MON *')

# Every 30 minutes
@schedule(cron='0/30 * * * ? *')

# Last day of every month at midnight
@schedule(cron='0 0 L * ? *')

# First Monday of every month
@schedule(cron='0 0 ? * MON#1 *')

Minute Hour Day Month Weekday

# Every day at 3 AM UTC
@schedule(cron='0 3 * * *')

# Every Monday at 9 AM
@schedule(cron='0 9 * * 1')

# Every 2 hours
@schedule(cron='0 */2 * * *')

from metaflow import FlowSpec, schedule, step

# Run at 9 AM Pacific Time
@schedule(cron='0 9 * * *', timezone='America/Los_Angeles')
class PacificTimeFlow(FlowSpec):
    
    @step
    def start(self):
        from datetime import datetime
        print(f"Current time: {datetime.now()}")
        self.next(self.end)
    
    @step
    def end(self):
        pass

# Deploy with schedule
python myflow.py step-functions create

# Deploy with Argo Workflows
python myflow.py argo-workflows create

# Manual trigger with Step Functions
python myflow.py step-functions trigger

# Manual trigger with Argo Workflows
python myflow.py argo-workflows trigger

kubectl patch cronworkflow <workflow-name> -p '{"spec":{"suspend":true}}'

# Change from daily to hourly
@schedule(hourly=True)  # was: @schedule(daily=True)
class MyFlow(FlowSpec):
    pass

python myflow.py step-functions create

from metaflow import FlowSpec, schedule, step, Parameter, resources

@schedule(hourly=True)
class HourlyETLFlow(FlowSpec):
    """ETL pipeline that runs every hour"""
    
    source = Parameter('source',
                      help='Data source to process',
                      default='s3://my-bucket/data/')
    
    @step
    def start(self):
        from datetime import datetime
        self.timestamp = datetime.utcnow().isoformat()
        print(f"Starting ETL at {self.timestamp}")
        self.next(self.extract)
    
    @resources(memory=4000)
    @step
    def extract(self):
        # Extract data from source
        print(f"Extracting data from {self.source}")
        self.data = self.load_data()
        self.next(self.transform)
    
    @resources(memory=8000, cpu=2)
    @step
    def transform(self):
        # Transform data
        self.processed_data = self.process(self.data)
        self.next(self.load)
    
    @step
    def load(self):
        # Load to destination
        self.save_results(self.processed_data)
        self.next(self.end)
    
    @step
    def end(self):
        print(f"ETL completed at {self.timestamp}")

from metaflow import FlowSpec, schedule, step, card

@schedule(cron='0 8 ? * MON *')  # Every Monday at 8 AM UTC
class WeeklyReportFlow(FlowSpec):
    """Generate weekly business reports"""
    
    @step
    def start(self):
        from datetime import datetime, timedelta
        self.end_date = datetime.utcnow()
        self.start_date = self.end_date - timedelta(days=7)
        self.next(self.fetch_data)
    
    @step
    def fetch_data(self):
        # Fetch last week's data
        self.metrics = self.get_weekly_metrics(
            self.start_date, 
            self.end_date
        )
        self.next(self.generate_report)
    
    @card
    @step
    def generate_report(self):
        # Generate visualizations and report
        from metaflow.cards import Markdown, Table
        
        report = Markdown(f"# Weekly Report\n\n"
                         f"Period: {self.start_date} to {self.end_date}")
        
        metrics_table = Table(self.metrics)
        
        self.next(self.send_report)
    
    @step
    def send_report(self):
        # Send report via email or Slack
        self.notify_stakeholders()
        self.next(self.end)
    
    @step
    def end(self):
        print("Weekly report sent successfully")

from metaflow import FlowSpec, schedule, step, project

# For global team: Run during business hours in multiple regions
@schedule(cron='0 9,14,22 * * *')  # 9 AM, 2 PM, 10 PM UTC
class GlobalSyncFlow(FlowSpec):
    """Sync data for teams across time zones"""
    
    @step
    def start(self):
        from datetime import datetime
        import pytz
        
        current_hour = datetime.utcnow().hour
        
        if current_hour == 9:
            self.region = 'APAC'
            self.tz = pytz.timezone('Asia/Tokyo')
        elif current_hour == 14:
            self.region = 'EMEA'
            self.tz = pytz.timezone('Europe/London')
        else:
            self.region = 'Americas'
            self.tz = pytz.timezone('America/New_York')
        
        self.next(self.sync_region)
    
    @step
    def sync_region(self):
        print(f"Syncing data for {self.region}")
        self.next(self.end)
    
    @step
    def end(self):
        pass