Créer un pipeline de traitement en flux continu à l'aide d'un modèle Dataflow

Ce guide de démarrage rapide explique comment créer un pipeline de traitement en flux continu à l'aide d'un modèle Dataflow fourni par Google. Plus précisément, nous utiliserons ici le modèle Pub/Sub vers BigQuery à titre d'exemple.

Le modèle Pub/Sub vers BigQuery est un pipeline de flux de données capable de lire des messages au format JSON à partir d'un sujet Pub/Sub et de les écrire dans une table BigQuery.

Pour obtenir des instructions détaillées sur cette tâche directement dans la console Google Cloud , cliquez sur Visite guidée :

Visite guidée

Avant de commencer

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Pub/Sub, and Resource Manager APIs.

    Enable the APIs

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Pub/Sub, and Resource Manager APIs.

    Enable the APIs

    8.
  8. Créez un bucket Cloud Storage :
    1. In the Google Cloud console, go to the Cloud Storage Buckets page.

      Go to Buckets

    2. Click Create.
    3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
      1. For Name your bucket, enter a unique bucket name. Don't include sensitive information in the bucket name, because the bucket namespace is global and publicly visible.
      2. In the Choose where to store your data section, do the following:
        1. Select a Location type.
        2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
        3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

          Set up cross-bucket replication

          1. In the Bucket menu, select a bucket.

          2. In the Replication settings section, click Configure to configure settings for the replication job.

            The Configure cross-bucket replication pane appears.

            • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click Add a prefix.
            • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
            • Click Done.
      3. In the Choose how to store your data section, do the following:
        1. In the Set a default class section, select the following: Standard.
        2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
      4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
      5. In the Choose how to protect object data section, do the following:
        • Select any of the options under Data protection that you want to set for your bucket.
          • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
          • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
          • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
            • To enable Object Retention Lock, click the Enable object retention checkbox.
            • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
        • To choose how your object data will be encrypted, expand the Data encryption section (), and select a Data encryption method.
    4. Click Create.
  9. Copiez les éléments suivants, car vous en aurez besoin dans une section ultérieure :
    • Le nom de votre bucket Cloud Storage.
    • L'ID de votre projet Google Cloud .

      Pour trouver cet ID, consultez la section Identifier des projets.

  10. Pour suivre la procédure de ce guide de démarrage rapide, votre compte utilisateur doit disposer du rôle Administrateur Dataflow et du rôle Utilisateur du compte de service. Le compte de service Compute Engine par défaut doit disposer des rôles Nœud de calcul Dataflow, Administrateur des objets de l'espace de stockage, Éditeur Pub/Sub, Éditeur de données BigQuery et Lecteur. Pour ajouter les rôles requis dans la console Google Cloud  :

    1. Accédez à la page IAM et sélectionnez votre projet.
      Accéder à IAM
    2. Sur la ligne contenant votre compte utilisateur, cliquez sur Modifier le compte principal. Cliquez sur Ajouter un autre rôle, puis ajoutez les rôles Administrateur Dataflow et Utilisateur du compte de service.
    3. Cliquez sur Enregistrer.
    4. Sur la ligne contenant le compte de service Compute Engine par défaut (PROJECT_NUMBER-compute@developer.gserviceaccount.com), cliquez sur Modifier le compte principal.
    5. Cliquez sur Ajouter un autre rôle, puis ajoutez les rôles suivants : Nœud de calcul Dataflow, Administrateur des objets Storage, Éditeur Pub/Sub, Éditeur de données BigQuery et Lecteur.

    6. Cliquez sur Enregistrer.

      Pour en savoir plus sur l'attribution de rôles, consultez la page Attribuer un rôle IAM à l'aide de la console.

  11. Par défaut, chaque nouveau projet démarre avec un réseau par défaut. Si le réseau par défaut de votre projet est désactivé ou a été supprimé, vous devez disposer d'un réseau dans votre projet pour lequel votre compte utilisateur dispose du Rôle d'utilisateur de réseau de Compute (roles/compute.networkUser ).

Créer un ensemble de données et une table BigQuery

Créez un ensemble de données et une table BigQuery selon le schéma approprié à votre sujet Pub/Sub à l'aide de la console Google Cloud .

Dans cet exemple, le nom de l'ensemble de données est taxirides et le nom de la table est realtime. Pour créer cet ensemble de données et cette table, procédez comme suit :

  1. Accédez à la page BigQuery.
    Accéder à BigQuery
  2. Dans le panneau Explorateur, à côté du projet dans lequel vous souhaitez créer l'ensemble de données, cliquez sur Afficher les actions, puis cliquez sur Créer un ensemble de données.
  3. Dans le panneau Créer un ensemble de données, procédez comme suit :
    1. Dans le champ ID de l'ensemble de données, saisissez taxirides. Les ID des ensembles de données sont uniques pour chaque projet Google Cloud .
    2. Pour Type d'emplacement, choisissez Multirégional, puis sélectionnez US (plusieurs régions aux États-Unis). Les ensembles de données publics sont stockés dans l'emplacement multirégional US. Par souci de simplicité, utilisez le même emplacement pour votre ensemble de données.
    3. Conservez les autres paramètres par défaut, puis cliquez sur Créer un ensemble de données.
  4. Dans le panneau Explorateur, développez votre projet.
  5. À côté de votre ensemble de données taxirides, cliquez sur Afficher les actions, puis sur Créer une table.
  6. Dans le panneau Créer une table, procédez comme suit :
    1. Dans la section Source, sous Créer une table à partir de, sélectionnez Table vide.
    2. Pour le champ Table de la section Destination, saisissez realtime.
    3. Dans la section Schéma, cliquez sur le bouton Modifier sous forme de texte et collez la définition de schéma suivante dans la zone : 
      ride_id:string,point_idx:integer,latitude:float,longitude:float,timestamp:timestamp,
meter_reading:float,meter_increment:float,ride_status:string,passenger_count:integer
    4. Dans la section Paramètres de partitionnement et de clustering, sous Partitionnement, sélectionnez le champ Code temporel.
  7. Ne modifiez aucun autre paramètre par défaut et cliquez sur Créer une table.

Exécuter le pipeline

Exécutez un pipeline de flux de données à l'aide du modèle Pub/Sub vers BigQuery fourni par Google. Le pipeline reçoit des données entrantes à partir du sujet d'entrée.

  1. Accédez à la page Tâches Dataflow.
    Accéder aux tâches
  2. Cliquez sur Créer une tâche à partir d'un modèle.
  3. Saisissez taxi-data comme nom de la tâche pour votre tâche Dataflow.
  4. Pour Modèle Dataflow, sélectionnez le modèle Pub/Sub vers BigQuery.
  5. Dans le champ Table BigQuery de sortie, saisissez ce qui suit : 
    PROJECT_ID:taxirides.realtime

    Remplacez PROJECT_ID par l'ID du projet dans lequel vous avez créé l'ensemble de données BigQuery.

  6. Dans la section Paramètres source facultatifs, cliquez sur Saisir le sujet manuellement pour le champ Sujet Pub/Sub d'entrée.
  7. Dans la boîte de dialogue, saisissez les éléments suivants comme Nom du sujet, puis cliquez sur Enregistrer : 
    projects/pubsub-public-data/topics/taxirides-realtime

    Ce sujet Pub/Sub disponible publiquement est basé sur l'ensemble de données ouvert de la NYC Taxi & Limousine Commission. Voici un exemple de message de cet article au format JSON :

    {
  "ride_id": "19c41fc4-e362-4be5-9d06-435a7dc9ba8e",
  "point_idx": 217,
  "latitude": 40.75399,
  "longitude": -73.96302,
  "timestamp": "2021-03-08T02:29:09.66644-05:00",
  "meter_reading": 6.293821,
  "meter_increment": 0.029003782,
  "ride_status": "enroute",
  "passenger_count": 1
}
  8. Dans le champ Emplacement temporaire, saisissez ce qui suit : 
    gs://BUCKET_NAME/temp/

    Remplacez BUCKET_NAME par le nom de votre bucket Cloud Storage. Le dossier temp stocke les fichiers temporaires, comme la tâche de pipeline en préproduction.

  9. Si votre projet ne possède pas de réseau par défaut, saisissez un réseau et un sous-réseau. Pour plus d'informations, consultez la section Spécifier un réseau et un sous-réseau.
  10. Cliquez sur Exécuter la tâche.

Afficher les résultats

Pour afficher les données écrites dans la table realtime, procédez comme suit :

  1. Accédez à la page BigQuery.

    Accéder à BigQuery

  2. Cliquez sur Saisir une nouvelle requête. Un onglet Éditeur s'affiche.

    SELECT * FROM `PROJECT_ID.taxirides.realtime`
WHERE `timestamp` > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
LIMIT 1000

    Remplacez PROJECT_ID par l'ID du projet dans lequel vous avez créé l'ensemble de données BigQuery. L'affichage des données dans votre tableau peut prendre jusqu'à cinq minutes.

  3. Cliquez sur Exécuter.

    La requête renvoie les lignes qui ont été ajoutées à votre table au cours des dernières 24 heures. Vous pouvez également exécuter des requêtes en SQL standard.

Effectuer un nettoyage

Pour éviter que les ressources utilisées dans cette démonstration soient facturées sur votre compte Google Cloud , procédez comme suit :

Supprimer le projet

Le moyen le plus simple d'empêcher la facturation est de supprimer le projet Google Cloud que vous avez créé pour ce guide de démarrage rapide.

  1. In the Google Cloud console, go to the Manage resources page.

    Go to Manage resources

  2. In the project list, select the project that you want to delete, and then click Delete.
  3. In the dialog, type the project ID, and then click Shut down to delete the project.

Supprimer les ressources individuelles

Si vous souhaitez conserver le projet Google Cloud que vous avez utilisé dans ce guide de démarrage rapide, supprimez les ressources individuelles :

  1. Accédez à la page Tâches Dataflow.
    Accéder aux tâches
  2. Sélectionnez votre tâche de traitement par flux dans la liste des tâches.
  3. Dans la barre de navigation, cliquez sur Arrêter.
  4. Dans la boîte de dialogue Arrêter la tâche, annulez ou drainez votre pipeline, puis cliquez sur Arrêter la tâche.
  5. Accédez à la page BigQuery.
    Accéder à BigQuery
  6. Sur le panneau Explorateur, développez votre projet.
  7. À côté de l'ensemble de données que vous souhaitez supprimer, cliquez sur Afficher les actions, puis sur Ouvrir.
  8. Dans le panneau des détails, cliquez sur Supprimer l'ensemble de données, puis suivez les instructions.
  9. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  10. Click the checkbox for the bucket that you want to delete.
  11. To delete the bucket, click Delete, and then follow the instructions.

Étapes suivantes