Utiliser le connecteur Cloud Storage avec Apache Spark

Ce tutoriel vous montre comment exécuter un exemple de code utilisant le connecteur Cloud Storage avec Apache Spark.

Objectifs

Écrire une tâche de décompte de mots simple Spark en Java, Scala ou Python, puis exécuter la tâche sur un cluster Dataproc.

Coûts

Dans ce document, vous utilisez les composants facturables de Google Cloudsuivants :

Compute Engine
Dataproc
Cloud Storage

Vous pouvez obtenir une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût.

Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai sans frais.

Avant de commencer

Exécutez les étapes ci-dessous pour préparer l'exécution du code dans ce tutoriel.

Configurer votre projet Si nécessaire, configurez un projet avec les API Dataproc, Compute Engine et Cloud Storage activées, et la Google Cloud CLI installée sur votre ordinateur local.
Créer un bucket Cloud Storage Vous aurez besoin d'un stockage Cloud Storage pour stocker les données du tutoriel. Si vous n'en avez pas, créez un bucket dans votre projet.
1. Dans la console Google Cloud , accédez à la page Buckets Cloud Storage.
  Accéder à la page "Buckets"
2. Cliquez sur Créer.
3. Sur la page Créer un bucket, saisissez les informations concernant votre bucket. Pour passer à l'étape suivante, cliquez sur Continuer.
  1. Dans la section Premiers pas, procédez comme suit :
    - Saisissez un nom unique qui répond aux exigences relatives aux noms des buckets.
    - Pour ajouter une étiquette de bucket, développez la section Étiquettes (), cliquez sur Ajouter une étiquette, puis spécifiez un élément key et value pour votre étiquette.
  2. Dans la section Choisir l'emplacement de stockage de vos données, procédez comme suit :
    1. Sélectionnez un type d'emplacement.
    2. Choisissez un emplacement où les données de votre bucket seront stockées de manière permanente dans le menu déroulant Type d'emplacement.
      - Si vous sélectionnez le type d'emplacement birégional, vous pouvez également choisir d'activer la réplication turbo à l'aide de la case à cocher correspondante.
    3. Pour configurer la réplication entre buckets, sélectionnez Ajouter une réplication entre buckets via le service de transfert de stockage et suivez ces étapes :
      Configurer la réplication entre buckets
      
      Dans le menu Bucket, sélectionnez un bucket.
      
      Dans la section Paramètres de réplication, cliquez sur Configurer pour configurer les paramètres du job de réplication.
      
      Le volet Configurer la réplication entre buckets s'affiche.
      
      Pour filtrer les objets à répliquer en fonction du préfixe de leur nom, saisissez le préfixe avec lequel vous souhaitez inclure ou exclure des objets, puis cliquez sur Ajouter un préfixe.
      
      Pour définir une classe de stockage pour les objets répliqués, sélectionnez-en une dans le menu Classe de stockage. Si vous ignorez cette étape, les objets répliqués utiliseront la classe de stockage par défaut du bucket de destination.
      
      Cliquez sur OK.
  3. Dans la section Choisir comment stocker vos données, procédez comme suit :
    1. Sélectionnez une classe de stockage par défaut pour le bucket ou classe automatique pour gérer automatiquement les classes de stockage des données de votre bucket.
    2. Pour activer l'espace de noms hiérarchique, dans la section Optimiser l'espace de stockage pour les charges de travail utilisant beaucoup de données, sélectionnez Activer l'espace de noms hiérarchique sur ce bucket.
      Remarque : Vous ne pouvez pas activer l'espace de noms hiérarchique dans des buckets existants.
  4. Dans la section Choisir comment contrôler l'accès aux objets, indiquez si votre bucket applique ou non la protection contre l'accès public et sélectionnez une méthode de contrôle des accès pour les objets de votre bucket.
    Remarque : Vous ne pouvez pas modifier le paramètre Empêcher l'accès public s'il est appliqué au niveau d'une règle d'administration.
  5. Dans la section Choisir comment protéger les données d'objet, procédez comme suit :
    - Sous Protection des données, sélectionnez les options que vous souhaitez définir pour votre bucket.
      - Pour activer la suppression réversible, cochez la case Règle de suppression réversible (pour la récupération de données), puis spécifiez le nombre de jours pendant lesquels vous souhaitez conserver les objets après leur suppression.
      - Pour configurer la gestion des versions d'objets, cochez la case Gestion des versions des objets (pour le contrôle des versions), puis spécifiez le nombre maximal de versions par objet et le nombre de jours après lesquels les versions obsolètes expirent.
      - Pour activer la règle de conservation sur les objets et les buckets, cochez la case Conservation (pour la conformité), puis procédez comme suit :
        
        Pour activer le verrou de conservation des objets, cochez la case Activer la conservation des objets.
        
        Pour activer le verrou de bucket, cochez la case Définir une règle de conservation du bucket, puis choisissez une unité de temps et une durée pour votre période de conservation.
    - Pour choisir comment vos données d'objet seront chiffrées, développez la section Chiffrement des données (), puis sélectionnez une méthode de chiffrement des données.
4. Cliquez sur Créer.
Définissez des variables d'environnement locales. Définissez des variables d'environnement sur votre ordinateur local. Définissez l'ID de votre projet Google Cloud et le nom du bucket Cloud Storage que vous utiliserez pour ce tutoriel. Indiquez également le nom et la région d'un cluster Dataproc existant ou nouveau. Vous pouvez créer un cluster à utiliser dans ce tutoriel à l'étape suivante.
```
PROJECT=project-id
```
```
BUCKET_NAME=bucket-name
```
```
CLUSTER=cluster-name
```
```
REGION=cluster-region Example: "us-central1"
```
Créer un cluster Dataproc Exécutez la commande ci-dessous pour créer un cluster Dataproc à nœud unique dans la zone Compute Engine spécifiée.
```
gcloud dataproc clusters create ${CLUSTER} \
    --project=${PROJECT} \
    --region=${REGION} \
    --single-node
```
La commande ci-dessus installe la version d'image de cluster par défaut. Vous pouvez utiliser l'option --image-version pour sélectionner une version d'image pour votre cluster. Chaque version d'image installe des versions spécifiques des composants de la bibliothèque Spark et Scala. Si vous préparez la tâche de décompte de mots Spark en Java ou Scala, vous référencez les versions Spark et Scala installées sur votre cluster lorsque vous préparez le package de la tâche.
Copier des données publiques dans votre bucket Cloud Storage. Copiez un extrait de texte Shakespeare public dans le dossier input de votre bucket Cloud Storage :
```
gcloud storage cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
Configurer un environnement de développement Java (Apache Maven), Scala (SBT) ou Python.
Utiliser Cloud Shell. Cloud Shell inclut les outils utilisés dans ce tutoriel, y compris Apache Maven, Python et Google Cloud CLI.

Préparer la tâche de décompte Spark

Sélectionnez un onglet ci-dessous pour suivre la procédure de préparation d'un package ou d'un fichier de tâche à envoyer au cluster. Vous pouvez préparer l'un des types de tâches suivants :

Tâche Spark en Java utilisant Apache Maven pour créer un package JAR
Tâche Spark en Scala utilisant SBT pour créer un package JAR
Tâche Spark en Python (PySpark)

Java

Copiez le fichier pom.xml sur votre ordinateur local. Le fichier pom.xml suivant spécifie les dépendances des bibliothèques Scala et Spark, auxquelles un champ d'application provided est attribué pour indiquer que le cluster Dataproc fournira ces bibliothèques lors de l'exécution. Le fichier pom.xml ne spécifie pas de dépendance Cloud Storage, car le connecteur met en œuvre l'interface HDFS standard. Lorsqu'une tâche Spark accède à des fichiers de cluster Cloud Storage (fichiers dont les URI commencent par gs:// ), le système utilise automatiquement le connecteur Cloud Storage pour accéder aux fichiers dans Cloud Storage.

Vérifiez la version de votre image de cluster. Remplacez les espaces réservés version dans le fichier pour afficher les versions des bibliothèques Spark et Scala utilisées par la version d'image de votre cluster. Notez que le numéro d'artefact spark-core_ correspond au numéro de version major.minor de Scala.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>dataproc.codelab</groupId>
  <artifactId>word-count</artifactId>
  <version>1.0</version>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>Scala version, for example, 2.11.8</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_Scala major.minor.version, for example, 2.11</artifactId>
      <version>Spark version, for example, 2.3.1</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>
</project>

Copiez le code WordCount.java indiqué ci-dessous sur votre ordinateur local.

Créez un ensemble de répertoires avec le chemin d'accès src/main/java/dataproc/codelab :
```
mkdir -p src/main/java/dataproc/codelab
```
Copiez WordCount.java sur votre ordinateur local dans src/main/java/dataproc/codelab :
```
cp WordCount.java src/main/java/dataproc/codelab
```

WordCount.java est une tâche Spark en Java qui lit les fichiers texte de Cloud Storage, effectue un décompte des mots, puis écrit les résultats des fichiers texte dans Cloud Storage.

package dataproc.codelab;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

public class WordCount {
  public static void main(String[] args) {
    if (args.length != 2) {
      throw new IllegalArgumentException("Exactly 2 arguments are required: <inputUri> <outputUri>");
    }
    String inputPath = args[0];
    String outputPath = args[1];
    JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("Word Count"));
    JavaRDD<String> lines = sparkContext.textFile(inputPath);
    JavaRDD<String> words = lines.flatMap(
        (String line) -> Arrays.asList(line.split(" ")).iterator()
    );
    JavaPairRDD<String, Integer> wordCounts = words.mapToPair(
        (String word) -> new Tuple2<>(word, 1)
    ).reduceByKey(
        (Integer count1, Integer count2) -> count1 + count2
    );
    wordCounts.saveAsTextFile(outputPath);
  }
}

Créez le package.
```
mvn clean package
```
Si la compilation réussit, un fichier target/word-count-1.0.jar est créé.

Entreposez le package sur Cloud Storage.

gcloud storage cp target/word-count-1.0.jar \
    gs://${BUCKET_NAME}/java/word-count-1.0.jar

Scala

Copiez le fichier build.sbt sur votre ordinateur local. Le fichier build.sbt suivant spécifie les dépendances des bibliothèques Scala et Spark, auxquelles un champ d'application provided est attribué pour indiquer que le cluster Dataproc fournira ces bibliothèques lors de l'exécution. Le fichier build.sbt ne spécifie pas de dépendance Cloud Storage, car le connecteur met en œuvre l'interface HDFS standard. Lorsqu'une tâche Spark accède à des fichiers de cluster Cloud Storage (fichiers dont les URI commencent par gs:// ), le système utilise automatiquement le connecteur Cloud Storage pour accéder aux fichiers dans Cloud Storage.
Vérifiez la version de votre image de cluster. Remplacez les espaces réservés version dans le fichier pour afficher les versions des bibliothèques Spark et Scala utilisées par la version d'image de votre cluster.
```
scalaVersion := "Scala version, for example, 2.11.8"

name := "word-count"
organization := "dataproc.codelab"
version := "1.0"

libraryDependencies ++= Seq(
  "org.scala-lang" % "scala-library" % scalaVersion.value % "provided",
  "org.apache.spark" %% "spark-core" % "Spark version, for example, 2.3.1" % "provided"
)
```

Copiez word-count.scala sur votre ordinateur local. Il s'agit d'une tâche Spark en Java qui lit les fichiers texte de Cloud Storage, effectue un décompte des mots, puis écrit les résultats des fichiers texte dans Cloud Storage.

package dataproc.codelab

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    if (args.length != 2) {
      throw new IllegalArgumentException(
          "Exactly 2 arguments are required: <inputPath> <outputPath>")
    }

    val inputPath = args(0)
    val outputPath = args(1)

    val sc = new SparkContext(new SparkConf().setAppName("Word Count"))
    val lines = sc.textFile(inputPath)
    val words = lines.flatMap(line => line.split(" "))
    val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
    wordCounts.saveAsTextFile(outputPath)
  }
}

Créez le package.
```
sbt clean package
```
Si la compilation réussit, un fichier target/scala-2.11/word-count_2.11-1.0.jar est créé.

Entreposez le package sur Cloud Storage.

gcloud storage cp target/scala-2.11/word-count_2.11-1.0.jar \
    gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar

Python

Copiez word-count.py sur votre ordinateur local. Il s'agit d'une tâche Spark en Python qui lit les fichiers texte de Cloud Storage, effectue un décompte des mots, puis écrit les résultats des fichiers texte dans Cloud Storage.

#!/usr/bin/env python

import pyspark
import sys

if len(sys.argv) != 3:
  raise Exception("Exactly 2 arguments are required: <inputUri> <outputUri>")

inputUri=sys.argv[1]
outputUri=sys.argv[2]

sc = pyspark.SparkContext()
lines = sc.textFile(sys.argv[1])
words = lines.flatMap(lambda line: line.split())
wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda count1, count2: count1 + count2)
wordCounts.saveAsTextFile(sys.argv[2])

Envoyer la tâche

Exécutez la commande gcloud suivante pour envoyer la tâche de décompte à votre cluster Dataproc.

Java

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/java/word-count-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Scala

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Python

gcloud dataproc jobs submit pyspark word-count.py \
    --cluster=${CLUSTER} \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Consulter le résultat

Une fois le job terminé, exécutez la commande suivante de gcloud CLI pour afficher le résultat du décompte.

gcloud storage cat gs://${BUCKET_NAME}/output/*

Le résultat du décompte doit ressembler à celui-ci :

(a,2)
(call,1)
(What's,1)
(sweet.,1)
(we,1)
(as,1)
(name?,1)
(any,1)
(other,1)
(rose,1)
(smell,1)
(name,1)
(would,1)
(in,1)
(which,1)
(That,1)
(By,1)

Effectuer un nettoyage

Une fois le tutoriel terminé, vous pouvez procéder au nettoyage des ressources que vous avez créées afin qu'elles ne soient plus comptabilisées dans votre quota et qu'elles ne vous soient plus facturées. Dans les sections suivantes, nous allons voir comment supprimer ou désactiver ces ressources.

Supprimer le projet

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

Pour supprimer le projet :

Attention : La suppression d'un projet aura les effets suivants :

Tout le contenu du projet est supprimé. Si vous avez utilisé un projet existant pour les tâches décrites dans ce document et que vous le supprimez, vous supprimerez également tout autre travail effectué dans le projet.
Les ID de projets personnalisés sont perdus. Lorsque vous avez créé ce projet, vous avez peut-être créé un ID de projet personnalisé que vous souhaitez utiliser à l'avenir. Pour conserver les URL qui utilisent l'ID de projet, telle qu'une URL appspot.com, supprimez les ressources sélectionnées dans le projet au lieu de supprimer l'ensemble du projet.

Si vous envisagez d'explorer plusieurs architectures, tutoriels et guides de démarrage rapide, réutiliser des projets peut vous aider à ne pas dépasser les limites de quotas des projets.

Dans la console Google Cloud , accédez à la page Gérer les ressources.
Accéder à la page "Gérer les ressources"
Dans la liste des projets, sélectionnez le projet que vous souhaitez supprimer, puis cliquez sur Supprimer.
Dans la boîte de dialogue, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer le projet.

Supprimer le cluster Dataproc

Au lieu de supprimer votre projet, vous souhaiterez peut-être simplement supprimer votre cluster au sein du projet.

Supprimer le bucket Cloud Storage

ConsoleGoogle Cloud

Dans la console Google Cloud , accédez à la page Buckets de Cloud Storage.
Accéder à la page "Buckets"
Cochez la case correspondant au bucket que vous souhaitez supprimer.
Pour supprimer le bucket, cliquez sur Supprimer , puis suivez les instructions.

Ligne de commande

gcloud storage buckets delete BUCKET_NAME

Étapes suivantes

Consultez Conseils pour régler des tâches Spark.

Utiliser le connecteur Cloud Storage avec Apache Spark Restez organisé à l'aide des collections Enregistrez et classez les contenus selon vos préférences.

Objectifs

Coûts

Avant de commencer

Configurer la réplication entre buckets

Préparer la tâche de décompte Spark

Java

Scala

Python

Envoyer la tâche

Java

Scala

Python

Consulter le résultat

Effectuer un nettoyage

Supprimer le projet

Supprimer le cluster Dataproc

Supprimer le bucket Cloud Storage

ConsoleGoogle Cloud

Ligne de commande

Étapes suivantes

Utiliser le connecteur Cloud Storage avec Apache Spark