Examples

from metaflow import FlowSpec, step, conda

class HousePricePredictionFlow(FlowSpec):
    
    @conda(libraries={'scikit-learn': '1.0.2', 'numpy': '1.21.0'})
    @step
    def start(self):
        """Load and prepare data"""
        import numpy as np
        from sklearn.datasets import fetch_california_housing
        
        data = fetch_california_housing()
        self.features = data.data
        self.labels = data.target
        self.next(self.train_models, foreach='params')
    
    @property
    def params(self):
        return [
            {'n_estimators': 50, 'learning_rate': 0.1, 'max_depth': 3},
            {'n_estimators': 100, 'learning_rate': 0.05, 'max_depth': 5},
            {'n_estimators': 200, 'learning_rate': 0.01, 'max_depth': 7},
        ]
    
    @conda(libraries={'scikit-learn': '1.0.2', 'numpy': '1.21.0'})
    @step
    def fit_gbrt_for_given_param(self):
        """Fit GBRT with given parameters"""
        from sklearn import ensemble
        from sklearn.model_selection import cross_val_score
        import numpy as np
        
        estimator = ensemble.GradientBoostingRegressor(
            n_estimators=self.input['n_estimators'],
            learning_rate=self.input['learning_rate'],
            max_depth=self.input['max_depth'],
            min_samples_split=2,
            loss='ls'
        )
        
        estimator.fit(self.features, self.labels)
        
        mses = cross_val_score(
            estimator, self.features, self.labels,
            cv=5, scoring='neg_mean_squared_error'
        )
        rmse = np.sqrt(-mses).mean()
        
        self.fit = dict(
            index=int(self.index),
            params=self.input,
            rmse=rmse,
            estimator=estimator
        )
        
        self.next(self.select_best_model)
    
    @step
    def select_best_model(self, inputs):
        """Select the best model based on RMSE"""
        self.best_model = min(inputs, key=lambda x: x.fit['rmse'])
        self.next(self.end)
    
    @step
    def end(self):
        """End of flow"""
        print(f"Best RMSE: {self.best_model.fit['rmse']}")

if __name__ == '__main__':
    HousePricePredictionFlow()

from metaflow import FlowSpec, step, conda

class DataProcessingFlow(FlowSpec):
    
    @conda(libraries={'pandas': '1.3.0'})
    @step
    def start(self):
        import pandas as pd
        self.data = pd.DataFrame({
            'value': [100, 200, 0, 300],
            'divisor': [10, 20, 0, 30]  # Note: zero divisor!
        })
        self.next(self.process)
    
    @conda(libraries={'pandas': '1.3.0'})
    @step
    def process(self):
        import pandas as pd
        # This will fail on the third row
        self.data['result'] = self.data['value'] / self.data['divisor']
        self.next(self.end)
    
    @step
    def end(self):
        print(self.data)

if __name__ == '__main__':
    DataProcessingFlow()

from metaflow import FlowSpec, step, conda

class ParallelProcessingFlow(FlowSpec):
    
    @step
    def start(self):
        self.datasets = ['dataset_a', 'dataset_b', 'dataset_c']
        self.next(self.process, foreach='datasets')
    
    @conda(libraries={'pandas': '1.3.0'})
    @step
    def process(self):
        import pandas as pd
        self.dataset_name = self.input
        # Process the dataset
        self.result = f"Processed {self.dataset_name}"
        self.next(self.join)
    
    @step
    def join(self, inputs):
        self.results = [inp.result for inp in inputs]
        self.next(self.end)
    
    @step
    def end(self):
        print(self.results)

if __name__ == '__main__':
    ParallelProcessingFlow()

from metaflow import FlowSpec, step, conda

class VideoProcessingFlow(FlowSpec):
    
    @conda(
        libraries={'opencv-python': '4.5.3', 'pillow': '8.3.0'},
        python='3.8'
    )
    @step
    def start(self):
        import cv2
        from PIL import Image
        import numpy as np
        
        # Process video frame
        self.frame = np.random.rand(480, 640, 3) * 255
        self.frame = self.frame.astype(np.uint8)
        
        self.next(self.analyze)
    
    @conda(
        libraries={
            'tensorflow': '2.6.0',
            'opencv-python': '4.5.3'
        },
        python='3.8'
    )
    @step
    def analyze(self):
        import tensorflow as tf
        import cv2
        
        # Run inference
        self.analysis_result = "Detected objects"
        self.next(self.end)
    
    @step
    def end(self):
        print(self.analysis_result)

if __name__ == '__main__':
    VideoProcessingFlow()