Ingérer et transformer des données à l'aide d'un flux de données

Un flux de données est un diagramme logique représentant le flux des données provenant des ressources de données sources, telles qu'une base de données ou un fichier plat, vers les ressources de données cibles, telles qu'un lac de données ou un entrepôt de données.

Le flux des données de la source vers la cible peut subir une série de transformations pour agréger, nettoyer et façonner les données. Les ingénieurs de données et les développeurs ETC peuvent ensuite analyser ou recueillir des informations et utiliser ces données pour prendre des décisions d'affaires efficaces.

Dans ce tutoriel, vous allez :

Créer un projet où vous pouvez enregistrer le flux de données.
Ajouter des opérateurs Source et sélectionner les entités de données à utiliser dans le flux de données.
Utiliser des opérateurs de mise en forme et appliquer des transformations.
Identifier la ressource de données cible pour charger les données.

Avant de commencer

Pour ingérer et transformer des données à l'aide d'un flux de données, vous devez disposer des éléments suivants :

Accès à un espace de travail d'intégration de données. Voir Se connecter au service d'intégration de données.
Ressources de données sources et cibles créées.
Autorisation PAR_MANAGE activée pour le seau temporaire.
```
allow any-user to manage buckets in compartment <compartment-name> where ALL {request.principal.type = 'disworkspace', request.principal.id = '<workspace-ocid>', request.permission = 'PAR_MANAGE'}
```
Les bases de données autonomes utilisent le service de stockage d'objets pour stocker temporairement les données et ont besoin de demandes préauthentifiées.

1. Création d'un projet et d'un flux de données

Dans le service d'intégration de données pour Oracle Cloud Infrastructure, les flux de données et les tâches ne peuvent être créés que dans un projet ou un dossier.

Pour créer un projet et un flux de données :

Dans la page d'accueil de l'espace de travail, sélectionnez Projets.
Si vous avez besoin d'aide pour trouver l'espace de travail du service d'intégration de données dans lequel travailler, voir Création et accès à un espace de travail.
Dans la page Projets, sélectionnez Créer un projet, puis Créer.
Dans la page Créer un projet, entrez DI_Lab pour Nom, puis sélectionnez Créer.

Après avoir créé un projet, vous créez un flux de données pour ingérer des données à partir de deux fichiers sources contenant des informations sur les clients (CUSTOMERS.json) et les commandes (REVENUE.csv).
Dans la page de détails du projet DI_Lab, sélectionnez Flux de données dans le sous-menu de gauche.
Si vous êtes dans la page Projets, sélectionnez DI_Lab pour ouvrir la page de détails du projet.
Sélectionnez Créer un flux de données.

Le concepteur de flux de données s'ouvre dans un nouvel onglet.
Dans le panneau Propriétés, pour Nom, entrez Load Customers and Revenue Data.

La valeur Identificateur est générée automatiquement en fonction de la valeur que vous entrez pour le nom du flux de données. Vous pouvez modifier la valeur générée, mais après avoir enregistré le flux de données, vous n'êtes pas autorisé à mettre à jour l'identificateur.
Sélectionnez Créer.

Le concepteur reste ouvert pour que vous puissiez continuer à apporter des modifications.

2. Ajout d'opérateurs Source

Vous ajoutez des opérateurs Source pour identifier les entités de données à utiliser pour le flux de données. Dans ce tutoriel, une entité de données représente une table de base de données.

À partir du panneau Opérateurs, déposez un opérateur Source sur le canevas.
Dans le canevas, sélectionnez SOURCE_1, s'il n'est pas déjà sélectionné.

Le panneau Propriétés affiche les détails de l'opérateur ciblé.
Dans l'onglet Détails du panneau Propriétés, cliquez sur Sélectionner à côté de chacune des options suivantes pour effectuer les sélections :
- Dans le champ Ressource de données, sélectionnez Data_Lake.
- For Connection, select Default connection.
- Pour Schéma, sélectionnez le compartiment, puis le seau. Aux fins de ce tutoriel, Object Storage sert de ressource de données source, c'est pourquoi vous sélectionnez le seau ici.
- Pour entité de données, sélectionnez Parcourir par nom, puis CUSTOMERS.json. For File type, select JSON.
Lorsque vous avez terminé les sélections pour SOURCE_1, le nom de l'opérateur devient CUSTOMERS_JSON, ce qui reflète la sélection de l'entité de données. Dans le champ Identificateur, renommez l'opérateur Source à CUSTOMERS.
Répétez les étapes 1 à 3 pour ajouter un deuxième opérateur Source avec les valeurs suivantes :
- Dans le champ Ressource de données, sélectionnez Data_Lake.
- For Connection, select Default Connection.
- Pour Schéma, sélectionnez le compartiment, puis le seau. Aux fins de ce tutoriel, Object Storage sert de ressource de données source, c'est pourquoi vous sélectionnez le seau ici.
- Dans le champ Entité de données, sélectionnez REVENUE.csv, puis CSV comme type de fichier. Acceptez les valeurs par défaut pour les éléments restants.
Lorsque vous avez terminé les sélections pour le deuxième opérateur Source, le nom de l'opérateur devient REVENUE_CSV, reflétant la sélection de l'entité de données. Dans le champ Identificateur, renommez l'opérateur Source à REVENUE.
(Facultatif) Cliquez sur Affecter un paramètre pour empêcher les détails de la source d'être liés au code compilé, lorsque vous publiez le flux de données.
(Facultatif) Dans l'onglet Attributs, vous pouvez voir les attributs de l'entité de données et leur appliquer des règles d'exclusion ou de renommage à partir de leur menu Actions respectif (trois points).
Vous pouvez également utiliser l'icône de filtre de la colonne Nom ou Type pour appliquer un ou plusieurs filtres aux attributs à exclure.
(Facultatif) Dans l'onglet Données, vous pouvez voir un échantillon des données de l'entité de données source et appliquer des transformations aux données ou sélectionner des attributs pour voir un profil des données.
(Facultatif) Dans l'onglet Validation, vous pouvez vérifier la présence d'avertissements ou d'erreurs dans la configuration des opérateurs Source.
Pour enregistrer le flux de données et continuer de le modifier, sélectionnez Enregistrer.

3. Filtrage et transformation des données

Filtrage des données

L'opérateur Filtrer produit un sous-ensemble des données provenant d'un opérateur en amont en fonction d'une condition.

À partir du panneau Opérateurs, déposez un opérateur Filtrer sur le canevas.
Connectez REVENUE à FILTER_1 :
- Placez le curseur sur REVENUE.
- Faites glisser le cercle de connecteur à côté de REVENUE.
- Déposez le cercle du connecteur sur FILTER_1.
Sélectionnez FILTER_1.
Dans le panneau Propriétés, sélectionnez Create (Créer) à côté de Filter condition (Condition de filtre).
Dans le panneau Créer une condition de filtre, entrez STA dans le champ Rechercher par nom.
Cliquez deux fois sur ORDER_STATUS ou déposez-le pour l'ajouter à l'éditeur de condition de filtre.
Dans l'éditeur de condition, entrez ='1-Booked'. La condition est donc la suivante :

FILTER_1.REVENUE_CSV.ORDER_STATUS='1-Booked'

Note

Pour éviter des problèmes avec les guillemets, ne faites pas de copie et de collage.
Sélectionnez Créer.
Dans le panneau Opérateurs, déposez un opérateur Filtre sur le canevas, puis placez-le après CUSTOMERS.
Connectez CUSTOMERS à FILTER_2 :
- Placez le curseur sur CUSTOMERS.
- Faites glisser le cercle des connecteurs à droite de CUSTOMERS.
- Déposez le cercle du connecteur sur FILTER_2.
Dans le panneau Propriétés pour FILTER_2, sélectionnez Créer à côté de Condition de filtre.
Dans le panneau Créer une condition de filtre, entrez COU dans le champ Rechercher par nom.
Cliquez deux fois sur COUNTRY_CODE pour l'ajouter à l'éditeur de condition.
Entrez ='US' pour que la condition soit la suivante :

FILTER_2.CUSTOMERS_JSON.COUNTRY_CODE='US'
Sélectionnez Créer.
Pour enregistrer le flux de données et continuer de le modifier, sélectionnez Enregistrer.

Transformation de données

À l'aide de l'explorateur de données, vous pouvez explorer un échantillon de données, vérifier les métadonnées de profilage et appliquer des transformations dans l'onglet Données du panneau Propriétés. Des opérateurs Expression sont ajoutés au canevas pour chaque transformation appliquée.

Dans le panneau Services pour FILTER_2, cliquez sur l'onglet Données.
Toutes les rangées de données et tous les attributs sont affichés. Vous pouvez utiliser la barre de défilement verticale pour faire défiler les rangées et la barre de défilement horizontale pour faire défiler les attributs.
Dans le champ Rechercher par modèle, entrez STATE*.
Le nombre d'attributs dans la table est filtré. Seuls les attributs correspondant au modèle sont affichés.
Sélectionnez le menu des transformations () pour FILTER_2.CUSTOMERS_JSON.STATE_PROVINCE, puis sélectionnez Changer la casse.
In the Change case dialog, from the Type menu, select Upper.
Ne cochez pas la case Conserver les attributs sources.
Laissez le nom tel quel.
Sélectionnez Appliquer.

Un opérateur Expression est ajouté au flux de données. Dans le panneau Propriétés, l'onglet Détails est maintenant ciblé, affichant les détails de l'opérateur Expression.

Dans la table Expressions, vous pouvez voir l'expression générée, UPPER(EXPRESSION_1.CUSTOMERS_JSON.STATE_PROVINCE).
Dans le panneau Propriétés de l'opérateur Expression, remplacez le nom du champ Identificateur par CHANGE_CASE.
Cliquez sur l'onglet Données, puis utilisez la barre de défilement horizontale pour aller à la fin.

CHANGE_CASE.STATE_PROVINCE est ajouté à la fin du jeu de données. Vous pouvez prévisualiser les données transformées pour CHANGE_CASE.STATE_PROVINCE dans l'onglet Données.
Dans le panneau Opérateurs, déposez l'opérateur Expression sur le canevas, puis placez-le après CHANGE_CASE.
Connectez CHANGE_CASE au nouvel opérateur EXPRESSION_1.
Dans le panneau Propriétés pour EXPRESSION_1, sélectionnez Ajouter une expression dans le tableau Expressions.
Dans le panneau Ajouter une expression :
1. Dans le champ Identificateur, renommez l'expression à FULLNAME.
2. Conservez la valeur VARCHAR pour Type de données.
3. Définissez la longueur à 200.
4. Sous Générateur d'expression, passez de la liste Entrant à la liste Fonctions.
5. Dans le champ Rechercher par nom, entrez CON. Ensuite, repérez CONCAT sous String.
6. Entrez CONCAT(CONCAT(EXPRESSION_1.CUSTOMERS_JSON.FIRST_NAME, ' '),EXPRESSION_1.CUSTOMERS_JSON.LAST_NAME)
  
  Vous pouvez également mettre en évidence les paramètres fictifs d'une fonction, puis double-cliquer ou déposer des attributs à partir de la liste Entrant pour créer une expression.
7. Sélectionnez Ajouter.
L'expression CONCAT est maintenant répertoriée dans la table Expressions pour l'opérateur EXPRESSION_1. Vous pouvez ajouter autant d'expressions que vous le souhaitez.
Pour enregistrer le flux de données et continuer de le modifier, sélectionnez Enregistrer.

4. Création de jointures de données

Après avoir appliqué des filtres et des transformations, vous pouvez joindre les entités de données sources à l'aide d'un identificateur de client unique, puis charger les données dans une entité de données cible.

Pour joindre les données de EXPRESSION_1 aux données de FILTER_1, déposez un opérateur Jointure du panneau Opérateurs sur le canevas, puis placez-le après EXPRESSION_1 et FILTER_1.
Connectez EXPRESSION_1 à JOIN_1. Connectez ensuite FILTER_1 à JOIN_1.
Avec JOIN_1 sélectionné, dans l'onglet Détails du panneau Propriétés, sélectionnez Créer à côté de Condition de jointure.
Dans le panneau Créer une condition de jointure, entrez CUST dans le champ Rechercher par nom.
Vous souhaitez joindre les entités en utilisant CUST_ID et CUST_KEY.
Dans l'éditeur, entrez JOIN_1_1.CUSTOMERS_JSON.CUST_ID=JOIN_1_2.REVENUE_CSV.CUST_KEY.
Sélectionnez Créer.
Pour enregistrer le flux de données et continuer de le modifier, sélectionnez Enregistrer.

5. Ajout d'un opérateur Cible

À partir du panneau Opérateurs, déposez un opérateur Cible sur le canevas.
Connectez JOIN_1 à TARGET_1.
Sélectionnez TARGET_1 sur le canevas, s'il n'est pas déjà sélectionné.
Dans l'onglet Détails du panneau Propriétés, ne cochez pas la case Créer une entité de données.
Cliquez ensuite sur Sélectionner pour chacune des options suivantes et effectuer les sélections pour une entité de données :
- Pour Ressource de données, sélectionnez Data_Warehouse.
- For Connection, select Default Connection.
- Pour Schéma, sélectionnez BETA.
- Pour entité de données, sélectionnez CUSTOMERS_TARGET.
Pour Emplacement temporaire, sélectionnez la ressource de données Data Lake, sa connexion par défaut et le compartiment. Ensuite, pour Schéma, sélectionnez le seau de stockage d'objets que vous avez utilisé pour importer des exemples de données.
Conservez la valeur Insert (Insérer) pour la stratégie d'intégration.
Pour vérifier le mappage des attributs, cliquez sur l'onglet Mappage.

Par défaut, tous les attributs sont mappés par nom.

Par exemple, CUST_ID de JOIN_1 est mappé à CUST_ID dans l'entité de données cible.
Pour mapper manuellement des attributs qui ne sont pas encore mappés, sélectionnez le menu Tout dans la table attributs cibles, puis sélectionnez Attributs non mappés.

Vous pouvez faire de même dans la table attributs sources (pour les champs entrants.
Supprimez maintenant FULLNAME des attributs sources vers FULL_NAME dans Attributs cibles.
Dans la table attributs cibles, sélectionnez Tous les attributs non mappés, puis Tous.

Tous les attributs sont maintenant mappés.
(Facultatif) Sélectionnez Actions pour vérifier les différentes options de mappage disponibles.
(Facultatif) Sélectionnez Voir les règles pour afficher les règles appliquées.
Pour enregistrer le flux de données et quitter le concepteur, sélectionnez Enregistrer et fermer.

Ressources supplémentaires

Pour en savoir plus, voir :

Étape suivante

Après avoir ingéré et transformé des données à l'aide d'un flux de données, créez une tâche d'intégration pour configurer et exécuter le flux de données.

Documentation sur Oracle Cloud Infrastructure