Use the Cloud Storage connector with Apache Spark

This tutorial show you how to run example code that uses the Cloud Storage connector with Apache Spark.

Objectives

Write a simple wordcount Spark job in Java, Scala, or Python, then run the job on a Dataproc cluster.

A Dataproc cluster is pre-installed with the Spark components needed for this tutorial.

Costs

This tutorial uses billable components of Google Cloud, including:

• Compute Engine
• Dataproc
• Cloud Storage

Use the Pricing Calculator to generate a cost estimate based on your projected usage. New Cloud Platform users might be eligible for a free trial.

Before you begin

Run the steps below to prepare to run the code in this tutorial.

1. Set up your project. If necessary, set up a project with the Dataproc, Compute Engine, and Cloud Storage APIs enabled and the Cloud SDK installed on your local machine.

   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.

      Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

      Go to project selector

   3. Make sure that billing is enabled for your Cloud project. Learn how to confirm that billing is enabled for your project.

   4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

      Enable the APIs

   5. Create a service account:

      1. In the Cloud Console, go to the Create service account page.

         Go to Create service account
      2. Select a project.

      3. In the Service account name field, enter a name. The Cloud Console fills in the Service account ID field based on this name.

         In the Service account description field, enter a description. For example, Service account for quickstart.

      4. Click Create and continue.

      5. Click the Select a role field.

         Under Quick access, click Basic, then click Owner.

         Note: The Role field affects which resources your service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.
      6. Click Continue.

      7. Click Done to finish creating the service account.

         Do not close your browser window. You will use it in the next step.

   6. Create a service account key:

      1. In the Cloud Console, click the email address for the service account that you created.
      2. Click Keys.
      3. Click Add key, then click Create new key.
      4. Click Create. A JSON key file is downloaded to your computer.
      5. Click Close.

   7. Set the environment variable GOOGLE_APPLICATION_CREDENTIALS to the path of the JSON file that contains your service account key. This variable only applies to your current shell session, so if you open a new session, set the variable again.

      Example: Linux or macOS

      export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

      Replace KEY_PATH with the path of the JSON file that contains your service account key.

      For example:

      export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

      Example: Windows

      For PowerShell:

      $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

      Replace KEY_PATH with the path of the JSON file that contains your service account key.

      For example:

      $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

      For command prompt:

      set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

      Replace KEY_PATH with the path of the JSON file that contains your service account key.

   8. Install and initialize the Cloud SDK.

2. Create a Cloud Storage bucket. You need a Cloud Storage to hold tutorial data. If you do not have one ready to use, create a new bucket in your project.

   1. In the Cloud Console, go to the Cloud Storage Browser page.

      Go to Browser

   2. Click Create bucket.
   3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
      • For Name your bucket, enter a name that meets the bucket naming requirements.
      • For Choose where to store your data, do the following:
        • Select a Location type option.
        • Select a Location option.
      • For Choose a default storage class for your data, select a storage class.
      • For Choose how to control access to objects, select an Access control option.
      • For Advanced settings (optional), specify an encryption method, a retention policy, or bucket labels.
   4. Click Create.

3. Set local environment variables. Set environment variables on your local machine. Set your Google Cloud project-id and the name of the Cloud Storage bucket you will use for this tutorial. Also provide the name and region of an existing or new Dataproc cluster. You can create a cluster to use in this tutorial in the next step.

   PROJECT=project-id
   

   BUCKET_NAME=bucket-name
   

   CLUSTER=cluster-name
   

   REGION=cluster-region Example: "us-central1"
   

4. Create a Dataproc cluster. Run the command, below, to create a single-node Dataproc cluster in the specified Compute Engine zone.

   gcloud dataproc clusters create ${CLUSTER} \
       --project=${PROJECT} \
       --region=${REGION} \
       --single-node
   

   The above command installs the default cluster image version. You can use the --image-version flag to select an image version for your cluster. Each image version installs specific versions of Spark and Scala library components. If you prepare the Spark wordcount job in Java or Scala, you will reference the Spark and Scala versions installed on your cluster when you prepare the job package.

5. Copy public data to your Cloud Storage bucket. Copy a public data Shakespeare text snippet into the input folder of your Cloud Storage bucket:

   gsutil cp gs://pub/shakespeare/rose.txt \
       gs://${BUCKET_NAME}/input/rose.txt
   

6. Set up a Java (Apache Maven), Scala (SBT), or Python development environment.

   Use Cloud Shell. Cloud Shell includes tools used in this tutorial, including Apache Maven, Python, and the gcloud and gsutil Cloud SDK command-line tools.

Prepare the Spark wordcount job

Select a tab, below, to follow the steps to prepare a job package or file to submit to your cluster. You can prepare one of the following job types;

• Spark job in Java using Apache Maven to build a JAR package
• Spark job in Scala using SBT to build a JAR package
• Spark job in Python (PySpark)

Java

1. Copy pom.xml file to your local machine. The following pom.xml file specifies Scala and Spark library dependencies, which are given a provided scope to indicate that the Dataproc cluster will provide these libraries at runtime. The pom.xml file does not specify a Cloud Storage dependency because the connector implements the standard HDFS interface. When a Spark job accesses Cloud Storage cluster files (files with URIs that start with gs://), the system automatically uses the Cloud Storage connector to access the files in Cloud Storage
   Check your cluster image version. Replace the version placeholders in the file to show the Spark and Scala library versions used by your cluster's image version. Note that the spark-core_ artifact number is the Scala major.minor version number.

   <?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>
   

2. Copy the WordCount.java code listed, below, to your local machine.
   1. Create a set of directories with the path src/main/java/dataproc/codelab:

      mkdir -p src/main/java/dataproc/codelab
      

   2. Copy WordCount.java to your local machine into src/main/java/dataproc/codelab:

      cp WordCount.java src/main/java/dataproc/codelab
      

   WordCount.java is a simple Spark job in Java that reads text files from Cloud Storage, performs a word count, then writes the text file results to 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);
     }
   }
   

3. Build the package.

   mvn clean package
   

   If the build is successful, a target/spark-with-gcs-1.0-SNAPSHOT.jar is created.
4. Stage the package to Cloud Storage.

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

Scala

1. Copy build.sbt file to your local machine. The following build.sbt file specifies Scala and Spark library dependencies, which are given a provided scope to indicate that the Dataproc cluster will provide these libraries at runtime. The build.sbt file does not specify a Cloud Storage dependency because the connector implements the standard HDFS interface. When a Spark job accesses Cloud Storage cluster files (files with URIs that start with gs://), the system automatically uses the Cloud Storage connector to access the files in Cloud Storage
   Check your cluster image verison. Replace the version placeholders in the file to show the Spark and Scala library versions used by your cluster's image version.

   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"
   )
   
   

2. Copy word-count.scala to your local machine. This is a simple Spark job in Java that reads text files from Cloud Storage, performs a word count, then writes the text file results to 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)
     }
   }
   
   

3. Build the package.

   sbt clean package
   

   If the build is successful, a target/scala-2.11/word-count_2.11-1.0.jar is created.
4. Stage the package to Cloud Storage.

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

Python

1. Copy word-count.py to your local machine. This is a simple Spark job in Python using PySpark that reads text files from Cloud Storage, performs a word count, then writes the text file results to 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])
   

Submit the job

Run the following gcloud command to submit the wordcount job to your Dataproc cluster.

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/

View the output

After the job finishes, run the following Cloud SDK gsutil command to view the wordcount output.

gsutil cat gs://${BUCKET_NAME}/output/*

The wordcount output should be similar to the following:

(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)

Clean up

After you finish the tutorial, you can clean up the resources that you created so that they stop using quota and incurring charges. The following sections describe how to delete or turn off these resources.

Deleting the project

The easiest way to eliminate billing is to delete the project that you created for the tutorial.

To delete the project:

Caution: Deleting a project has the following effects:

• Everything in the project is deleted. If you used an existing project for this tutorial, when you delete it, you also delete any other work you've done in the project.
• Custom project IDs are lost. When you created this project, you might have created a custom project ID that you want to use in the future. To preserve the URLs that use the project ID, such as an appspot.com URL, delete selected resources inside the project instead of deleting the whole project.

If you plan to explore multiple tutorials and quickstarts, reusing projects can help you avoid exceeding project quota limits.

1. In the 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.

Deleting the Dataproc cluster

Instead of deleting your project, you may wish to only delete your cluster within the project.

Deleting the Cloud Storage bucket

Cloud Console

1. In the Cloud Console, go to the Cloud Storage Browser page.

   Go to Browser

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

Command line

Delete the bucket:

gsutil rb BUCKET_NAME

Important: Your bucket must be empty before you can delete it.

What's next

• See Spark job tuning tips