Avant de commencer
-
Vous avez précédemment créé une connexion au système stockant vos données source.
-
Vous avez précédemment ajouté le jeu de données contenant vos données source.
Ici, des données hiérarchiques concernant des taxis, notamment les heures de prise en charge et de fin de course, les tarifs, etc. (téléchargez le fichier type_converter-taxi.json depuis l'onglet Téléchargements dans le panneau de gauche de la page).
-
Vous avez créé la connexion et le jeu de données associé qui contiendra les données traitées.
Ici, un fichier stocké dans HDFS.
Procédure
-
Cliquez sur Add pipeline (Ajouter un pipeline) dans la page Pipelines. Votre nouveau pipeline s’ouvre.
-
Donnez-lui un nom significatif.
Exemple
Convert small taxi rides (Convertir les petites courses de taxi)
-
Cliquez sur ADD SOURCE pour ouvrir le panneau vous permettant de sélectionner vos données source, ici les données relatives aux taxis.
Avertissement : Le processeur Type converter ne peut traiter de sous-enregistrement. Si vous souhaitez convertir ces enregistrements, vous devez utiliser un processeur Field selector avant, afin de réorganiser les enregistrements et de les mettre en haut du schéma.
-
Sélectionnez votre jeu de données et cliquez sur Select (Sélectionner) pour l'ajouter au pipeline.
Renommez-le si nécessaire.
-
Cliquez sur le bouton
et ajoutez un processeur Field selector au pipeline. Le panneau de configuration s'ouvre.
-
Donnez un nom significatif au processeur.
Exemple
reorganize records
-
Cliquez sur l'icône
dans le mode de sélection Simple :
-
Sélectionnez le champ .pickup.pickup_datetime et renommez-le pickup_time, puisque vous souhaitez sélectionner et renommer le champ pickup_datetime du premier emplacement, puis le déplacer en haut du schéma.
-
Sélectionnez le champ .dropoff.dropoff_datetime et renommez-le dropoff_time, puisque vous souhaitez sélectionner et renommer le champ dropoff_datetime du premier emplacement, puis le déplacer en haut du schéma.
-
Sélectionnez le champ .payment.fare_amount et renommez-le fare, puisque vous souhaitez sélectionner et renommer le champ fare_amount du premier emplacement, puis le déplacer en haut du schéma.
-
Cliquez sur Edit (Modifier) puis sur Save (Sauvegarder) pour sauvegarder votre configuration.
-
Cliquez sur le bouton et ajoutez un processeur Type converter au pipeline. Le panneau de configuration s'ouvre.
-
Donnez un nom significatif au processeur.
Exemple
convert rides and fares
-
Dans la zone Converters :
-
Sélectionnez .pickup_time dans la liste Field path (Chemin du champ), sélectionnez le mode Primitive (Primitif), sélectionnez DateTime dans la liste Output type (Type de sortie) et saisissez yyyy-MM-dd HH:mm:ss dans le champ Format, car vous souhaitez convertir le champ de type DateTime contenant les heures de prise en charge en un champ de type Integer. yyyy-MM-dd HH:mm:ss correspond au format du champ d'entrée.
-
Cliquez sur l'icône + pour ajouter un convertisseur et sélectionnez .dropoff_time dans la liste Field path (Chemin du champ). Sélectionnez le mode Primitive (Primitif), sélectionnez DateTime dans la liste Output type (Type de sortie) et saisissez yyyy-MM-dd HH:mm:ss dans le champ Format, car vous souhaitez convertir le champ de type DateTime contenant les heures de fin de course en un champ de type Integer. yyyy-MM-dd HH:mm:ss correspond au format du champ d'entrée.
-
Cliquez sur l'icône + pour ajouter un convertisseur et sélectionnez .fare dans la liste Field path (Chemin du champ). Sélectionnez le mode Primitive (Primitif) et sélectionnez Double dans la liste Output type (Type de sortie), puisque vous souhaitez convertir le champ de type String contenant le prix des courses en un champ de type Double.
Conseil : Vous avez la possibilité d'appliquer plusieurs conversions sur le même champ. Par exemple, vous pouvez convertir un champ de type String contenant une date en un champ de type Long et utiliser ce champ généré de type Long pour le convertir en un champ de type DateTime.
-
Cliquez sur Save (Sauvegarder) pour sauvegarder votre configuration.
-
Cliquez sur après le processeur Type Converter dans le pipeline et ajoutez un processeur Filter.
-
Donnez un nom significatif au processeur.
Exemple
filter on short rides
-
Dans la zone Filters (Filtres) :
-
Saisissez .{.dropoff_time - .pickup_time > 660000} dans la liste Input, pour filtrer les courses ayant duré moins de 11 minutes.
-
Sélectionnez Count (Nombre) dans la liste Optionally select a function to apply (De manière facultative, sélectionnez une fonction à appliquer), > dans la liste Operator (Opérateur) et saisissez 0 dans la liste Value (Valeur) puisque vous souhaitez compter ces courses rapides.
-
Cliquez sur Save (Sauvegarder) pour sauvegarder votre configuration.
-
(Facultatif) Consultez l'aperçu du processeur Filter pour voir les données après l'opération de filtre.
-
Cliquez sur ADD DESTINATION (AJOUTER UNE DESTINATION) dans le pipeline pour ouvrir le panneau vous permettant de sélectionner le jeu de données qui contiendra vos données (HDFS).
-
Donnez-lui un nom significatif, short rides data par exemple.
-
Dans la barre d'outils en haut de Talend Cloud Pipeline Designer, cliquez sur le bouton Run (Exécuter) pour ouvrir le panneau vous permettant de sélectionner votre profil d'exécution.
-
Sélectionnez dans la liste votre profil d'exécution (pour plus d'informations, consultez Profils d'exécution), puis cliquez sur Run (Exécuter) pour exécuter votre pipeline.
Résultats
Votre pipeline est en cours d'exécution, les types de champs sont convertis et filtrés et le flux de sortie est envoyé vers le système cible défini.
Avertissement : Le processeur Type converter ne peut traiter de sous-enregistrement. Si vous souhaitez convertir ces enregistrements, vous devez utiliser un processeur Field selector avant, afin de réorganiser les enregistrements et de les mettre en haut du schéma.
