Apprenez Metaflow en 10 minutes

Metaflow est un framework Python créé dans Netflix et axé sur le domaine de la science des données. À savoir, il est conçu pour créer des projets visant à travailler avec des données et pour gérer de tels projets. Récemment, la société l'a transféré dans la catégorie open-source. Le framework Metaflow a été largement utilisé au sein de Netflix au cours des 2 dernières années. En particulier, il a permis de réduire considérablement le temps nécessaire à la conclusion des projets en production.



Le matériel que nous traduisons aujourd'hui est un guide rapide de Metaflow.

Qu'est-ce que Metaflow?

Vous trouverez ci-dessous un graphique illustrant la mise en œuvre du cadre Metaflow dans Netflix.


Implémentation de Metaflow dans Netflix

En novembre 2018, ce cadre a été utilisé dans 134 projets de l'entreprise.

Metaflow est un cadre pour créer et exécuter des workflows de science des données. Il présente les fonctionnalités suivantes:

• Gestion des ressources informatiques.
• Lancement de tâche conteneurisée.
• Gestion des dépendances externes.
• Gestion des versions, réexécution des tâches, poursuite de l'exécution des tâches suspendues.
• API client pour examiner les résultats des tâches qui peuvent être utilisées dans l'environnement Jupyter Notebook.
• Prise en charge de l'exécution de tâches locales (par exemple, sur un ordinateur portable) et à distance (dans le cloud). Possibilité de basculer entre ces modes.

L'utilisateur vtuulos a écrit sur Ycombinator que Metaflow peut automatiquement créer des instantanés (instantanés) de code, de données et de dépendances. Tout cela est placé dans un référentiel avec adressage par contenu, qui est généralement basé sur S3, bien que le système de fichiers local soit également pris en charge. Cela vous permet de continuer à exécuter des tâches arrêtées, de reproduire les résultats précédemment obtenus et d'explorer tout ce qui concerne les tâches, par exemple dans le bloc-notes Jupyter.

En général, nous pouvons dire que Metaflow vise à augmenter la productivité des scientifiques de données. Cela est dû au fait que le cadre leur permet de s'engager exclusivement dans le travail avec des données, sans être distrait par la résolution de tâches connexes. De plus, Metaflow accélère le retrait des projets basés sur celui-ci en production.


Les besoins d'un data scientist liés à ses responsabilités directes et la solution des tâches auxiliaires liées à l'infrastructure sur laquelle les calculs sont effectués

Scénarios de workflow avec Metaflow

Voici quelques scénarios de workflow que vous pouvez organiser à l'aide de Metaflow:

• Collaboration Un data scientist veut aider un autre à trouver la source de l'erreur. Dans le même temps, l'assistant souhaite télécharger sur son ordinateur l'ensemble de l'environnement dans lequel la tâche qui s'est écrasée a fonctionné.
• Poursuite des tâches arrêtées depuis l'endroit où elles ont été arrêtées. Une tâche s'est arrêtée avec une erreur (ou a été arrêtée intentionnellement). L'erreur a été corrigée (ou le code a été modifié). Il est nécessaire de redémarrer la tâche pour que son travail continue à partir de l'endroit où elle a échoué (ou a été arrêtée).
• Exécution de tâches hybrides. Vous devez effectuer une certaine étape du flux de travail localement (il s'agit peut-être de télécharger des données à partir d'un fichier stocké dans un dossier sur l'ordinateur), et une autre étape qui nécessite de grandes ressources de calcul (il s'agit peut-être de la formation du modèle) doit être effectuée dans le cloud.
• Examen des métadonnées obtenues après avoir terminé une tâche. Trois scientifiques des données sont engagés dans la sélection d'hyperparamètres du même modèle, en essayant d'améliorer la précision de ce modèle. Après cela, vous devez analyser les résultats de l'exécution des tâches de formation du modèle et sélectionner l'ensemble d'hyperparamètres qui s'est avéré être le meilleur.
• Utilisation de plusieurs versions du même package. Dans le projet, vous devez utiliser différentes versions, par exemple, les bibliothèques sklearn. Pendant le prétraitement, sa version 0.20 est requise et pendant la modélisation, la version 0.22 est requise.

Flux de travail typique de Metaflow

Considérez un flux de travail Metaflow typique d'un point de vue conceptuel et de programmation.

LookAvis conceptuel sur le flux de travail Metaflow

D'un point de vue conceptuel, les workflows Metaflow (chaînes de tâches) sont représentés par des graphiques acycliques dirigés (DAG). Les illustrations ci-dessous vous aideront à mieux comprendre cette idée.


Graphique acyclique linéaire


Graphique acyclique avec chemins "parallèles"

Chaque nœud du graphique représente une étape de traitement des données dans le flux de travail.

À chaque étape de la chaîne de tâches, Metaflow exécute du code Python normal sans aucune modification spéciale. Le code est exécuté dans des conteneurs séparés dans lesquels le code est compressé avec ses dépendances.

Un aspect clé de l'architecture Metaflow est représenté par le fait qu'il vous permet d'implémenter presque toutes les bibliothèques externes de l'écosystème conda dans des projets basés sur celui-ci sans utiliser de plugins. Cela distingue Metaflow des autres solutions polyvalentes similaires. Par exemple - depuis Airflow.

▍ Flux de travail Metaflow en termes de programmation

Chaque chaîne de tâches (flux) peut être représentée comme une classe Python standard (les noms de ces classes ont généralement le mot Flow ) si elle satisfait aux exigences minimales suivantes:

• La classe est la descendante de la classe Metaflow FlowSpec .
• Le @step chaque fonction qui représente une étape de la chaîne de tâches.
• À la fin de chaque fonction @step , il doit y avoir une indication d'une fonction similaire qui la suit. Cela peut être fait en utilisant une construction de ce type: self.next(self.function_name_here) .
• La classe implémente les fonctions de start et de end .

Prenons un exemple de chaîne minimale de tâches composée de trois nœuds.

Son schéma ressemble à ceci:

 start → process_message → end 

Voici son code:

 from metaflow import FlowSpec, step class LinearFlow(FlowSpec):         """     ,      Metaflow.    """       #         @step    def start(self):        self.message = 'Thanks for reading.'        self.next(self.process_message)    @step    def process_message(self):        print('the message is: %s' % self.message)        self.next(self.end)    @step    def end(self):        print('the message is still: %s' % self.message) if __name__ == '__main__':    LinearFlow() 

Instructions d'installation de Metaflow

▍Installation et essai

Voici la séquence d'étapes que vous devez effectuer pour installer et lancer d'abord Metaflow:

• Installer Metaflow (Python 3 recommandé): pip3 install metaflow .
• Mettez le fragment de code ci-dessus ( ici sur GitHub) dans le fichier linear_flow.py .
• Pour consulter l'architecture de la chaîne de tâches implémentée par ce code, utilisez la commande python3 linear_flow.py show .
• Pour démarrer le flux, exécutez la python3 linear_flow.py run .

Vous devriez obtenir quelque chose de similaire à celui illustré ci-dessous.


Bilan de santé Metaflow réussi

Ici, il convient de prêter attention à certaines choses. Le framework Metaflow crée un .metaflow données .metaflow local. Là, il stocke toutes les métadonnées liées à l'exécution des tâches et les instantanés associés aux sessions d'exécution des tâches. Si vous avez configuré des paramètres Metaflow liés au stockage de données dans le cloud, les instantanés seront stockés dans AWS S3 Bucket et les métadonnées liées aux lancements de tâches iront au service de métadonnées basé sur RDS (Relational Data Store, Relational Data Store). Plus tard, nous parlerons de la façon d'explorer ces métadonnées à l'aide de l'API client. Une autre bagatelle, bien qu'importante, à laquelle il convient de prêter attention, est que les identificateurs de processus (pid, ID de processus) attachés aux différentes étapes diffèrent. N'oubliez pas - nous avons dit ci-dessus que Metaflow conteneurise indépendamment chaque étape de la chaîne de tâches et effectue chaque étape dans son propre environnement (en ne transmettant que les données entre les étapes).

▍Installation et configuration de conda (si vous prévoyez d'implémenter des dépendances)

Suivez ces étapes pour installer conda:

• Téléchargez et installez Miniconda.
• Ajoutez un canal conda avec la commande conda config --add channels conda-forge .

Vous êtes maintenant prêt à intégrer des dépendances conda dans vos chaînes de tâches. Les détails de ce processus seront discutés ci-dessous.

Exemple de workflow réaliste

Ci-dessus, nous avons expliqué comment installer Metaflow et comment vous assurer que le système est opérationnel. De plus, nous avons discuté des bases de l'architecture de workflow et regardé un exemple simple. Nous examinons ici un exemple plus complexe, tout en révélant certains des concepts de Metaflow.

▍Emploi

Créez un workflow à l'aide de Metaflow qui implémente les fonctions suivantes:

• Chargement de données de film CSV dans une trame de données Pandas.
• Calcul parallèle des quartiles pour les genres.
• Enregistrement d'un dictionnaire avec les résultats des calculs.

▍ Chaîne de tâches

Le squelette de la classe GenreStatsFlow est illustré GenreStatsFlow - GenreStatsFlow . Après l'avoir analysé, vous comprendrez l'essence de l'approche mise en œuvre ici pour résoudre notre problème.

 from metaflow import FlowSpec, step, catch, retry, IncludeFile, Parameter class GenreStatsFlow(FlowSpec):  """    ,  ,   .         :    1)  CSV-   Pandas.    2)     .    3)     .  """   @step  def start(self):    """         :        1)      Pandas.        2)    .        3)        .    """       # TODO:  CSV         self.genres = []    self.next(self.compute_statistics, foreach='genres') #  1     @catch(var='compute_failed') #  2  @retry(times=1) #  3  @step  def compute_statistics(self):    """    .   ."""    self.genre = self.input #  4    # TODO:        self.next(self.join)     @step  def join(self, inputs):    """       ."""    # TODO:      self.next(self.end)     @step  def end(self):      """End the flow."""      pass   if __name__ == '__main__':  GenreStatsFlow() 

Considérez certaines parties importantes de cet exemple. Le code contient des commentaires de la forme # n , auxquels nous ferons référence ci-dessous.

• Dans le 1 , à l'étape de start , faites attention au paramètre foreach . Grâce à lui, des copies des étapes compute_statistics sont compute_statistics en compute_statistics dans une boucle for each entrée de la liste des genres .
• Dans le 2 décorateur @catch(var='compute_failed') interceptera toute exception qui compute_statistics étape compute_statistics et l'écrira dans la variable compute_failed (il peut être lu à l'étape suivante).
• Dans le 3 décorateur @retry(times=1) fait exactement ce que son nom laisse @retry(times=1) . À savoir, lorsque des erreurs se produisent, il répète l'étape.
• D'où vient le 4 , dans compute_statistics , self.input ? Le fait est que l' input est une variable de classe fournie par Metaflow. Il contient des données applicables à une instance particulière de compute_statistics (lorsqu'il existe plusieurs copies d'une fonction exécutée en parallèle). Cette variable n'est ajoutée par Metaflow que lorsque les nœuds sont représentés par plusieurs processus parallèles, ou lorsque plusieurs nœuds sont combinés.
• Voici un exemple d'exécution de la même fonction en parallèle - compute_statistics . Mais, si nécessaire, vous pouvez simultanément exécuter des fonctions complètement différentes qui ne sont pas liées les unes aux autres. Pour ce faire, changez ce qui est montré dans le 1 en quelque chose comme self.next(self.func1, self.function2, self.function3) . Bien sûr, avec cette approche, il sera également nécessaire de réécrire l'étape de join , ce qui permettra de traiter les résultats des différentes fonctions sur celle-ci.

Voici comment imaginer la classe squelette ci-dessus.


Représentation visuelle de la classe GenreStatsFlow

▍Lisez le fichier de données et les paramètres de transfert

• Téléchargez ce fichier CSV de film.
• Vous devez maintenant équiper le programme de la prise en charge de la possibilité de transférer dynamiquement le chemin d'accès au fichier movie_data et la valeur max_genres à la max_genres . Le mécanisme des arguments externes nous y aidera. Metaflow vous permet de passer des arguments au programme en utilisant des indicateurs supplémentaires dans la commande de démarrage du workflow. Par exemple, cela pourrait ressembler à ceci: python3 tutorial_flow.py run --movie_data=path/to/movies.csv --max_genres=5 .
• Metaflow fournit au développeur des objets IncludeFile et Parameter qui vous permettent de lire l'entrée dans le code de workflow. Nous nous référons aux arguments passés en affectant des objets IncludeFile et Parameter aux variables de classe. Cela dépend de ce que nous voulons lire exactement - le fichier ou la valeur habituelle.

Voici à quoi ressemble le code en lisant les paramètres passés au programme lors de son lancement à partir de la ligne de commande:

     movie_data = IncludeFile("movie_data",                             help="The path to a movie metadata file.",                             default = 'movies.csv')                               max_genres = Parameter('max_genres',                help="The max number of genres to return statistics for",                default=5) 

▍Inclusion de conda dans la chaîne de tâches

• Si vous n'avez pas encore installé conda, reportez-vous à la section sur l'installation et la configuration de conda dans cet article.
• Ajoutez le décorateur GenreStatsFlow fourni par Metaflow à la classe GenreStatsFlow. Ce décorateur s'attend à recevoir la version python. Il peut être soit défini dans le code, soit obtenu à l'aide d'une fonction auxiliaire. Vous trouverez ci-dessous le code qui illustre l'utilisation du décorateur et montre une fonction d'assistance.
  
  

   def get_python_version():    """     ,    python,       .         conda        python.    """    import platform    versions = {'2' : '2.7.15',                '3' : '3.7.4'}    return versions[platform.python_version_tuple()[0]] #       python. @conda_base(python=get_python_version()) class GenreStatsFlow(FlowSpec): 

• Vous pouvez maintenant ajouter le décorateur @conda à n'importe quelle étape de la chaîne de tâches. Il attend un objet avec des dépendances, qui lui est transmis via le paramètre de libraries . Metaflow, avant de commencer l'étape, se chargera de préparer le conteneur avec les dépendances spécifiées. Si nécessaire, vous pouvez utiliser en toute sécurité différentes versions de packages à différentes étapes, car Metaflow lance chaque étape dans un conteneur séparé.
  
  

       @conda(libraries={'pandas' : '0.24.2'})    @step    def start(self): 

• Exécutez maintenant la commande suivante: python3 tutorial_flow.py --environment=conda run .

StartDébut de l'implémentation

 @conda(libraries={'pandas' : '0.24.2'})    @step    def start(self):    """         :        1)      Pandas.        2)    .        3)        .    """        import pandas        from io import StringIO        #      Pandas.        self.dataframe = pandas.read_csv(StringIO(self.movie_data))        #   'genres'      .         #   .        self.genres = {genre for genres \                       in self.dataframe['genres'] \                       for genre in genres.split('|')}        self.genres = list(self.genres)        #        .        #  'foreach'             #          self.next(self.compute_statistics, foreach='genres') 

Considérez certaines des fonctionnalités de ce code:

• Notez que l'expression d'importation pandas se trouve à l'intérieur de la fonction qui décrit l'étape. Le fait est que cette dépendance n'est introduite par conda que dans le cadre de cette étape.
• Mais les variables déclarées ici ( dataframe et genres ) sont disponibles même dans le code des étapes effectuées après cette étape. Le fait est que Metaflow fonctionne sur la base des principes de séparation des environnements d'exécution de code, mais permet aux données de se déplacer naturellement entre les étapes de la chaîne de tâches.

▍ Implémentation de l'étape compute_statistics

 @catch(var='compute_failed')    @retry    @conda(libraries={'pandas' : '0.25.3'})    @step    def compute_statistics(self):        """            .        """        #             # 'input'.        self.genre = self.input        print("Computing statistics for %s" % self.genre)        #         ,         #        .        selector = self.dataframe['genres'].\                   apply(lambda row: self.genre in row)        self.dataframe = self.dataframe[selector]        self.dataframe = self.dataframe[['movie_title', 'genres', 'gross']]        #     gross   .        points = [.25, .5, .75]        self.quartiles = self.dataframe['gross'].quantile(points).values        #  ,    .        self.next(self.join) 

Veuillez noter que dans cette étape, nous nous référons à la variable de dataframe qui a été déclarée à l'étape de start précédente. Nous modifions cette variable. Lorsque vous passez aux étapes suivantes, cette approche, qui implique l'utilisation d'un nouvel objet de dataframe modifié, vous permet d'organiser un travail efficace avec les données.

▍ Mettre en œuvre l'étape de jointure

 @conda(libraries={'pandas' : '0.25.3'})    @step    def join(self, inputs):        """               .        """        inputs = inputs[0:self.max_genres]        #   ,    .        self.genre_stats = {inp.genre.lower(): \                            {'quartiles': inp.quartiles,                             'dataframe': inp.dataframe} \                            for inp in inputs}        self.next(self.end) 

Ici, il convient de souligner quelques points:

• Dans cette étape, nous utilisons une version complètement différente de la bibliothèque pandas.
• Chaque élément du tableau d' inputs est une copie des compute_statistics précédemment exécutées. Il contient l'état de la fonction correspondante exécutée, c'est-à-dire les valeurs de diverses variables. Ainsi, l' input[0].quartiles peut contenir des quartiles pour le genre comedy , et l' input[1].quartiles input[0].quartiles peut contenir des quartiles pour le genre sci-fi .

EadyProjet prêt

Le code de projet complet que nous venons de revoir se trouve ici .

Pour voir comment fonctionne le flux de travail décrit dans le fichier tutorial_flow.py , vous devez exécuter la commande suivante:

 python3 tutorial_flow.py --environment=conda show 

Utilisez la commande suivante pour démarrer le workflow:

 python3 tutorial_flow.py --environment=conda run --movie_data=path/to/movies.csv --max_genres=7 

Examen des résultats de l'exécution d'un flux de travail à l'aide de l'API client

Afin d'examiner des instantanés de données et l'état des lancements précédents du flux de travail, vous pouvez utiliser l' API client fournie par Metaflow. Cette API est idéale pour explorer les détails des expériences effectuées dans l'environnement Jupyter Notebook.

Voici un exemple simple de la sortie de la variable genre_stats , tirée des données du dernier lancement réussi de GenreStatsFlow .

 from metaflow import Flow, get_metadata #      print("Using metadata provider: %s" % get_metadata()) #     MovieStatsFlow. run = Flow('GenreStatsFlow').latest_successful_run print("Using analysis from '%s'" % str(run)) genre_stats = run.data.genre_stats print(genre_stats) 

Exécution de workflows dans le cloud

Après avoir créé et testé le flux de travail sur un ordinateur ordinaire, il est très probable que vous souhaitiez exécuter le code dans le cloud pour accélérer le travail.

Actuellement, Metaflow prend uniquement en charge l'intégration avec AWS. Dans l'image suivante, vous pouvez voir un mappage des ressources locales et cloud utilisées par Metaflow.


Intégration de Metaflow et AWS

Pour connecter Metaflow à AWS, vous devez effectuer les étapes suivantes:

• Vous devez d'abord effectuer une configuration AWS unique en créant des ressources avec lesquelles Metaflow peut travailler. Les mêmes ressources peuvent être utilisées, par exemple, par des membres d'une équipe de travail qui se démontrent mutuellement les résultats des workflows. Vous pouvez trouver des instructions pertinentes ici. Les paramètres sont assez rapides, car Metaflow dispose d'un modèle de paramètres CloudFormation.
• Ensuite, sur l'ordinateur local, vous devez exécuter la metaflow configure aws et entrer les réponses aux questions système. Avec ces données, Metaflow pourra utiliser des entrepôts de données basés sur le cloud.
• Maintenant, pour démarrer des workflows locaux dans le cloud, ajoutez simplement la clé de --with batch à la --with batch démarrage du workflow. Par exemple, cela pourrait ressembler à ceci: python3 sample_flow.py run --with batch .
• Pour effectuer un lancement hybride du flux de travail, c'est-à-dire pour effectuer certaines étapes localement et d'autres dans le cloud, vous devez ajouter le décorateur @batch aux étapes qui doivent être effectuées dans le cloud. Par exemple, comme ceci: @batch(cpu=1, memory=500) .

Résumé

Ici, je voudrais noter quelques fonctionnalités de Metaflow qui peuvent être considérées à la fois comme les avantages et les inconvénients de ce cadre:

• Metaflow est étroitement intégré à AWS. Mais dans les plans de développement du cadre, il existe un support pour un plus grand nombre de fournisseurs de cloud.
• Metaflow est un outil qui ne prend en charge que l'interface de ligne de commande. Il ne possède pas d'interface graphique (contrairement à d'autres cadres universels pour organiser les processus de travail, tels que Airflow).

Chers lecteurs! Envisagez-vous d'utiliser Metaflow?