Introduction

Le but de cet article est de présenter l’outil DataFlow dans Azure Data Factory (ADF), depuis son intégration dans des pipelines ADF jusqu’à son monitoring, en passant par l’utilisation plus ou moins avancée d’activités. Cette présentation se basera sur le cas d’usage de lecture de fichiers, nettoyage et mise à jour d’une table delta. Cela permettra de comparer les dataflows avec d’autres alternatives comme Databricks par exemple. Cet article complète l’utilisation d’ADF présenté dans l’article suivant : https://blog.atawiz.fr/article/data-ingestion-adf qui ne couvrait pas les dataflow, uniquement les pipelines et leur monitoring.

Utilisation basique

Prérequis

Afin de reproduire les dataflows détaillés plus tard, il est nécessaire d’avoir les ressources suivantes :

• Un Azure Storage Account (pour les sources et les destinations des données)
• Un Azure Data Factory
• Connaitre les bases de ADF

Présentation de l’outil

Contexte

Un dataflow est une activité de ADF et comme pour les activités standards comme la copie, on a besoin d’au moins une source de données en entrée (source) et en sortie (sink). Un data flow est constitué d’un enchainement d’activités, comme peut l’être un pipeline ADF. Ces activités permettent d’accéder et de manipuler les données des datasets. On peut par exemple filtrer des données, faire des jointures entre plusieurs datasets, faire des agrégations ou pivoter des données, puis les écrire dans un autre dataset.

Activités – Cas basique

Pour créer un dataflow, il faut avoir au moins un dataset en entrée et un dataset en sortie, possiblement le même. Ici on crée :

• Un dataset pointant vers un dossier du datalake contenant des JSON qui sera notre source de données
• Un dossier contenant des CSV pour stocker les résultats. L’objectif est de lire des JSON qui ont cette structure et de les transformer pour obtenir des CSV sous cette forme.

On crée un dataflow en précisant le dataset en source. Dans notre cas, nos fichiers JSON sources contiennent des tableaux d’éléments ayant 4 propriétés.

On ajoute une activité pour flatten le JSON afin d’avoir 1 ligne par élément du tableau, puis on utilise l’activité derivedColumn pour créer 1 colonne par propriété.

On filtre ensuite sur les données qui nous intéresse, ici les lignes dont la valeur est supérieure ou égale à 0. On a accès au panneau d’« expression builder » sur les champs de la plupart des activités, et on peut y introduire des fonctions, références à des colonnes, variables, paramètres etc. de la même façon que sur les activités des pipelines.

Pour terminer, on insert nos données dans le dataset de destination avec l’activité « sink ». Par défaut les fichiers CSV de destination auront des noms générés automatiquement. Il peut être utile de forcer un pattern ou un nom de fichier en dur dans le cas d’un fichier CSV unique en sortie. Dans ce dernier, il faut prendre garde au fait que cela impose un traitement avec 1 seule partition, ce qui peu ralentir le traitement car pas de parallélisation. On peut également ajouter un dossier dans lequel ADF stockera les exceptions qui ont été levées lors des traitements (via l’onglet « Errors »).

Preview/Debug

Afin d’avoir une idée des résultats que chacune des activités va engendrer, on peut utiliser le mode preview. En l’activant on a le choix du TTL de la session et des paramètres à utiliser pour la session si des paramètres ont été ajoutés. Il est conseillé d’ajouter des paramètres car des comportements indésirables peuvent survenir lors de projections sur des sources / destinations de fichiers si les paramètres ne sont pas remplis.

En l’activant, on peut demander à avoir en preview les données à chacune des étapes pour vérifier que les transformations demandées donnent le bon résultat. Cette preview n’est pas mise à jour automatiquement, et on doit cliquer sur « Refresh » après une modification pour voir la dite preview s’actualiser (dans l’onglet « Data preview »). On voit ici que notre filtre ne laisse que les lignes désirées.

Une fois confiant dans le dataflow créé on peut le valider et le rattacher à un pipeline ADF.

Intégration avec pipeline et monitoring

Pour qu’un dataflow s’exécute il faut qu’il soit orchestré par ADF, au même titre que les autres activités, dans un pipeline. On crée un pipeline et on choisit une activité de dataflow. On sélectionne notre dataflow, éventuellement les paramètres déclarés dans notre dataflow. On peut ensuite choisir de compléter le pipeline avec d’autres activités, comme une copie des fichiers CSV contenant le résultat de notre dataflow, vers un autre dossier du datalake.

Une fois le pipeline créé, on peut l’exécuter (via un trigger normal ou en mode debug) et consulter son déroulement dans le panneau de monitoring. Nous n’avons pas plus d’informations sur le traitement de l’activité dataflow au niveau du pipeline, mais en cliquant sur le détail de l’activité on a une vue plus précise. Sur la page de monitoring d’un datafow, en cliquant sur « Stages » d’un sink, on a un aperçu du nombre de lignes traitées à chaque étape ainsi que le temps de traitement. Ces informations peuvent être utiles pour débugger des filtres, jointures ou agrégations par exemple, ainsi qu’optimiser les temps de traitement en identifiant les activités les plus chronophages.

En cliquant sur une des activités du dataflow, nous avons plus d’informations sur le nombre de données ayant transité par cette activité. On a le détail des partitions utilisées, et la répartition des données avec l’asymétrie (skewness) et l’aplatissement (kurtosis).

Flowlets et scripts

Certains traitements nécessitent des opérations plus complexes sur les données que permettent l’utilisation basique des activités.

Création et utilisation de flowlets

Lors des usages un peu plus complexes des dataflows, le nombre d’activités peut être démultiplié et le suivi des tâches devient plus difficile à suivre. Précédemment nous avons transformé un JSON en CSV en éclatant les propriétés en colonnes et en filtrant sur l’une d’entre elle. Dans un cas plus réaliste, on pourrait avoir en entrée un fichier JSON et un fichier de configuration pour enrichir les données à l’aide d’une jointure, puis upsert les données dans le CSV de destination avec une gestion en Slowly Changing Dimesion (SCD) de type 3 par exemple. Un flowlet consiste en une succession d’activités, utilisables dans des dataflows comme une simple activité. Les intérêts principaux sont de pouvoir éviter la duplication de code en factorisant des activités réutilisées dans des flowlets (évite la duplication de code pour une meilleure maintenabilité), et profiter d’une meilleure lisibilité du dataflow et de son monitoring.

Classique

On peut créer un flowlet en passant par le même menu que les dataflow. Contrairement aux dataflow, on a la possibilité d’avoir en entrée des inputs en plus des sources. Les sources fonctionnent de la même façon, tandis que les inputs sont l’équivalent des paramètres à passer à une sous fonction. Ici on définit un input avec des colonnes de type string, avec les mêmes noms que celles des propriétés du fichier JSON.

On reproduit les mêmes activités que sur le dataflow et on conclut le flowlet par une activité d’output. Comme pour toutes les activités de destination, on a la possibilité de choisir de mapper les colonnes automatiquement, ou une à une si les noms des colonnes du dataset d’entrée ne correspond pas exactement à celles de la destination.

Pour utiliser ce flowlet à partir d’un dataflow, on crée une activité de flowlet dans laquelle on associe le flowlet souhaité. Le dataset d’input configuré dans le flowlet doit correspondre avec celui passé dans le dataflow.

Rapide

Une autre façon de créer des flowlets existe, souvent plus rapide et plus pratique que la façon manuelle, car plus automatique et évite donc les erreurs de typages. On part d’activité d’un dataflow ou d’un flowlet existant, et en les sélectionnant à la souris avec la touche CTRL, puis en faisant clic droit, on a la possibilité de crée un flowlet à partir de notre sélection. Cela va créer à la fois l’enchaînement d’activités sélectionnées avec les configurations faite dans le dataflow, mais également l’input d’entrée avec le typage des colonnes.

Limites des flowlets

Sous-flowlets

L’utilisation de flowlet n’est pas aussi flexible que la création de sous fonctions. En effet il n’est pas possible d’utiliser de flowlet dans un flowlet. Il n’est donc pas possible de créer de nombreux flowlets, responsables chacun d’un traitement précis, et d’avoir des flowlets les regroupant en macro-traitements et enfin des dataflows orchestrant ces différents traitements.

Plusieurs sources

Il n’est pas possible d'utiliser plusieurs inputs dans un même flowlet, pour y faire une "join" ou un "exists" par exemple. Donc dans le cas où on doit réaliser ce genre d'opérations de multiples fois dans nos dataflows, il n'est pas possible de les factoriser dans des flowlets.

###Monitoring

Lors de monitoring de dataflow utilisant des flowlet, on peut observer des informations quantitatives et qualitatives sur chacune des activités, excepté sur le flowlet. On ne peut pas non plus monitorer les activités à l’intérieur du flowlet. On est donc restreint à analyser le dataflow à partir des activités placées directement à la racine du dataflow.

Utilisation des scripts

Présentation

Sur ADF, la visualisation principale / par défaut des assets est celui de l’interface no code, où la création des activités se fait en glisser déposer. On a aussi accès à la représentation sous format JSON de tous les assets et c’est sous cette forme qu’est stocké et traité l‘ensemble des traitements et ressources d’ADF. Cette interface plus technique est utilisable directement par l’utilisateur mais au niveau des pipelines il est souvent plus difficile de l’utiliser par rapport à l’interface de base. En revanche pour les dataflows et flowlets, une 3ème interface existe : à mi-chemin entre les 2 autres, il s’agit de l’interface ‘Script’. Dans cette interface, les traitements sont représentés sous la forme de code. On peut retrouver ce code dans l’interface JSON mais stringifié et plus difficilement lisible et manipulable.

Réplication

Cette interface Script est utile pour avoir d’un seul coup d’œil la totalité des traitements du dataflow, de pouvoir effectuer des recherches avec CTRL+F et renommer ou modifier des traitements comme on le souhaite sans les contraintes de l’interface par défaut. Avec cette interface, il est possible de copier totalement ou partiellement un script en un simple CTRL+C et de le répliquer dans un autre dataflow, en étant certain que les configurations soient les mêmes.

Utilisation avancée des activités

L’utilisation des scripts permet de faire les mêmes traitements que l’interface par défaut, mais sous la forme de code. Une fois la syntaxe du langage comprise, il est plus simple de réaliser des traitements complexes directement sous la forme de code plutôt que de trouver la combinaison de boutons, d’onglets et de valeurs à remplir dans les activités de l’interface de base.

Par exemple, on souhaite faire une unicité sur 3 colonnes d’un dataset, et ne garder qu’une occurrence (la première) pour ne pas avoir de doublons sur ces propriétés. Si on regarde au niveau du script, on analyse rapidement ce qui est fait. On crée une nouvelle colonne en hashant les 3 colonnes voulues, on fait un group by dessus et si on match (si on retrouve le même hash sur une autre ligne, et donc la même combinaison des 3 colonnes), on ne garde que la première occurrence avec la fonction first.

Si on regarde maintenant l’interface par défaut équivalente à cette activité, on retrouve nos morceaux de code, mais éclatés entre plusieurs onglets. De plus, pour réaliser cette activité via cette interface, il faut aussi maitriser le code, bien qu’on puisse en partie s’aider du panneau d’« expression builder » pour remplir les champs.

#Dataflow et table delta

Cas pratique

Il existe néanmoins des limites à l’utilisation des dataflows. Pour en présenter certaines, on se basera sur un exemple plus concret de merge de données sous format delta. On utilise le format delta car c’est un format mis en avant par Databricks, et par Microsoft récemment en intégrant de base ce format dans OneLake, le OneDrive des données utilisées par Fabric. Le dataflow a pour but de lire des données d’une table delta de Silver (architecture médaillon) et d’enrichir le dataset à l’aide d’une autre table delta via une jointure, et d’écrire le résultat dans une table delta Gold. Des filtres sont également ajoutés sur les dates pour réduire la taille des datasets d’entrée et réduire le temps de traitement de la jointure.

Limite des merge pour les inline datasets

Les sources et destinations des dataflows ne permettent pas d’utiliser les tables delta comme datasets. Il faut utiliser le type de source « Inline », et spécifier le chemin où est stockée la table. Il est ensuite nécessaire d’importer le schéma de la table via le panneau « Projection ». Plus haut dans l’article, lorsqu’on utilisait directement de dataset cette étape n’était pas obligatoire car le schéma était intégré au dataset.

Au niveau du Sink, il est possible de ne pas importer le schéma en choisissant l’automapping. En revanche si le nom et types des colonnes du dataset ne correspond pas à ceux de la table delta de destination, le dataflow ne fonctionnera pas. Il est souvent plus prudent de passer par un mapping fixe pour identifier les possibles erreurs. Avant de pouvoir faire une opération sur une table delta, il est également nécessaire de créer une activité d’« alter row » pour chacune des opérations souhaitées. Il n’est par exemple pas possible d’update des lignes sous certaines conditions, et de faire un insert en même temps dans le Sink, il faut faire 2 branches avec un alter row de chaque côté avant d’écrire en 2 temps dans le Sink.

Limite du debug

Lorsque le mode debug est activé, un sampling est fait pour éviter de travailler avec trop de données. Cette limite peut être modifiée mais il existe un plafond de plusieurs milliers ou dizaines de milliers de lignes. Il y a également une limite de visualisation de 100 lignes sur la partie Data preview des activités. Ces limites sont souvent suffisantes mais dans le cas de transformations nécessitant la totalité des données (agrégation ou jointure par exemple), on peut obtenir des résultats incomplets, rendant impossible le travail en mode preview. Dans le cas où on a des datasets de plusieurs millions de lignes, le mode preview de dataflow prendra parmi ces lignes uniquement quelques milliers, c'est à dire environ 1%. Mais lors d'une jointure entre de si petits échantillons, il est possible que le résultat en mode debug soit un dataset vide. Or un dataset vide dans la partie Data preview ne permet plus de vérifier les résultats des activités suivantes.

Comparaison avec Databricks

Dans l’optique d’effectuer des modifications sur des données, une alternative à l’utilisation de dataflow est l’utilisation de notebooks Databricks. De plus cette alternative est particulièrement adaptée lorsque les données sont stockées en format delta.

Il est possible d’intégrer l’exécution d’un notebook comme activité dans un pipeline ADF. Mais afin de profiter de toutes les possibilités de l’outil, il est nécessaire de migrer également la partie orchestration dans Databricks. Ce n’est pas toujours envisageable, notamment en raison d’un coût plus élevé lors de simple copie ou déplacement de fichiers. On peut avoir un temps de démarrage de cluster Databricks plus long que le warmup des dataflows, ce qui peut nuire à certains cas d’usage où le temps d’exécution total du pipeline est prioritaire. En revanche les possibilités sont bien plus grandes sur un notebook, ayant à disposition tout l’écosystème Spark et même Python/Scala/SQL.

Conclusion

L’utilisation de dataflow sur ADF semble donc être à privilégier pour de la manipulation de données dans des cas relativement simples. Les dataflows ont le mérite d’être intégrés à ADF ce qui permet de restreindre la totalité de la chaîne à 1 seul outil. Une utilisation plus complexe, bien que possible à l’aide des scripts et des flowlets, tend à rendre la solution difficilement maintenable et/ou supervisable. Des alternatives existent, comme l’utilisation de notebooks Databricks. Cependant la contrainte de création, migration de nouvelles ressources dans d’autres langages, avec le coût que ces changements impliquent, peut être un frein à ces solutions.