Filtrage, fusion et ajout d’une nouvelle colonne dans Azure Data Factory

Usine de données Azure est un outil ETL puissant, avec la capacité de créer des pipelines ETL en utilisant une approche low code/no code. Ceci peut être réalisé en utilisant « Activités”. Les activités sont les tâches exécutées sur les données d’un pipeline.

Dans cet article, je démontre un processus ETL qui copie des données d’une source à une autre et effectue certaines tâches sur les données à l’aide de Activités dans Azure Data Factory comme suit :

1. Extrait les fichiers avec une sous-chaîne spécifique dans leurs noms de fichier à partir d’un pool de fichiers de données comme source,
2. les fusionne en un seul fichier – les fichiers de données ont le même schéma – et,
3. Ajoute une colonne « Created Datetime » au fichier de données fusionnées.

Marchons à travers les étapes.

1. Extraction de fichiers avec une sous-chaîne spécifique dans leurs noms de fichiers

Selon la structure de la source, extraire les fichiers requis de la source peut être très simple, comme extraire tous les fichiers d’un dossier ou tous les fichiers avec une extension spécifique (*.txt ou *.csv) dans un dossier . Cependant, ce n’est pas toujours le cas. Supposons que vous ayez un dossier avec une variété de types de fichiers et de noms de fichiers comme source et que vous deviez sélectionner manuellement des fichiers spécifiques pour votre entrée de pipeline. La figure ci-dessous montre un exemple d’un tel dossier qui contient plusieurs fichiers csv et txt. Cependant, je ne suis intéressé que par l’extraction des fichiers csv qui ont « mycsv » et la date d’aujourd’hui 20220721, date à laquelle j’écris ce blog, dans leurs noms de fichiers (deux fichiers en surbrillance).

Quelqu’un peut suggérer d’utiliser « Filtrer par dernière modification » de l’activité de copie, car nous recherchons les fichiers d’aujourd’hui.

Mais ces fichiers sont tous chargés quotidiennement, et cette option n’est pas utile car elle extrait des données non pertinentes dans le processus ETL.

Pour avoir un filtrage approprié sur les fichiers d’entrée dans ce cas, nous utilisons Filtre activité sous Itération et conditions, qui fournit un outil pour filtrer les variables d’entrée en fonction d’une condition définie par l’utilisateur. Voyons comment.

Tout d’abord, nous devons obtenir les métadonnées du contenu du dossier pour obtenir les noms de fichiers, en utilisant Obtenir des métadonnées activité sous Général.

Dans Réglages onglet, après avoir défini un jeu de données pour Obtenir des métadonnées activité, j’ajoute un Liste de champ pour Articles enfants. En sélectionnant Articles enfants, Obtenir des métadonnées L’activité renvoie le nom et le type du contenu du dossier, illustré ci-dessous.

Après avoir collecté les métadonnées du dossier source, je dois extraire uniquement les fichiers avec la sous-chaîne ‘mycsv’ dans leurs noms de fichiers, en utilisant Filtre activité. Lors de la création d’un Filtreil y a deux champs qui doivent être fournis dans Réglages languette:

• Articles: Le tableau des éléments qui doivent être vérifiés en entrée.
• Condition: La condition de filtrage définie par l’utilisateur qui doit être vérifiée pour les tableaux d’entrée.

Pour notre propos, Articles est la sortie de Obtenir des métadonnées activité, et Condition spécifie la sous-chaîne d’intérêt, comme suit :

• Articles: @activity(‘Obtenir les métadonnées’).output.childItems
• Condition: @contains(item().name,’mycsv’)

Comme indiqué ci-dessous, seuls les fichiers avec « mycsv » dans leur nom ont passé le filtre.

Ensuite, en ajoutant un deuxième filtre, j’extrait les fichiers avec la date d’aujourd’hui, 20220721, dans leurs noms de fichiers. L’entrée du second filtre est la sortie des premiers, des fichiers avec ‘mycsv’ dans leurs noms. Articles et Condition sont définis comme suit :

• Articles: @activity(‘Filtrer les fichiers mycsv’).output.value
• Condition: @contains(item().name,utcNow(‘yyyyMMdd’))

Noter: Ces filtres de remorquage peuvent être combinés en un seul filtre comme suit :

• Articles: @activity(‘Obtenir les métadonnées’).output.childItems
• Condition: @and(contains(item().name,’mycsv’),contains(item().name,utcNow(‘yyyyMMdd’)))

2. Fusionner des fichiers avec le même schéma dans un seul fichier, en utilisant ADF Copy Activity

Ensuite, je fusionne les fichiers filtrés à l’étape précédente en un seul fichier en utilisant Copie activité. Ce n’est faisable que lorsque les fichiers ont le même schéma.

Dans Copie activité, Jeu de données source est le dossier qui contient les fichiers filtrés.

Dans Évier onglet, je définis le jeu de données du récepteur qui contient le fichier fusionné et définit le Comportement de copie à Fusionner des fichiers. Cela combine les fichiers et crée un seul fichier dans le dossier de destination.

3. Ajout d’une nouvelle colonne « Created Datetime » au fichier fusionné, à l’aide d’ADF

Après avoir fusionné des fichiers en un seul fichier, j’aimerais capturer la date et l’heure d’exécution du pipeline, en ajoutant une nouvelle colonne « Create Datetime » dans CST au fichier fusionné. Ainsi, dans La source onglet de Copie activité, tout en bas, il y a une option « Colonnes supplémentaires » qui permet de définir une nouvelle colonne pour le jeu de données avant de l’exporter vers la destination. Alors, je clique Nouveauajouter Nom comme « Créer Datetime » et Évaluer (ajouter une valeur dynamique) à @convertFromUtc(utcNow(), ‘Central Standard Time’).

Utilisant Cartographie tab, je peux mapper la nouvelle colonne sur n’importe quelle colonne existante dans la table de destination, si nécessaire.