This tutorial show you how to run example code that uses the [Cloud Storage connector](https://docs.cloud.google.com/managed-spark/docs/concepts/connectors/cloud-storage#clusters) with [Apache Spark](https://spark.apache.org "Apache Spark").
[Lightning Engine](https://cloud.google.com/products/lightning-engine?e=48754) enhances connectivity to Cloud Storage to optimize the performance of its native engine. The improved Cloud Storage connector minimizes metadata operations to reduce costs, while an optimized file output committer unlocks performance and reliability for Spark workloads. Complete the [Early Access](https://docs.google.com/forms/d/e/1FAIpQLSdkNW13znxnwskYuEgZ-BWLmyxbwg4aMnAU-Z2f3FGzYhHHmg/viewform?usp=header) form to request early access to this private preview feature. ## Objectives Write a simple wordcount Spark job in Java, Scala, or Python, then run the job on a Managed Service for Apache Spark cluster.
> [!IMPORTANT] > All Managed Service for Apache Spark cluster image versions have the Spark components needed for this tutorial already installed. ## Costs In this document, you use the following billable components of Google Cloud: - Compute Engine - Managed Service for Apache Spark - Cloud Storage To generate a cost estimate based on your projected usage, use the [pricing calculator](https://docs.cloud.google.com/products/calculator). New Google Cloud users might be eligible for a [free trial](https://docs.cloud.google.com/free).
## 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 Managed Service for Apache Spark, Compute Engine, and Cloud Storage APIs enabled and the Google Cloud CLI installed on your local machine.
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 Google Cloud console, go to the Cloud Storage **Buckets** page. [Go to Buckets](https://console.cloud.google.com/storage/browser) 2. Click **Create**. 3. On the **Create a bucket** page, enter your bucket information. To go to the next step, click **Continue** . 1. In the **Get started** section, do the following: - Enter a globally unique name that meets the [bucket naming requirements](https://docs.cloud.google.com/storage/docs/bucket-naming#requirements). - To add a [bucket label](https://docs.cloud.google.com/storage/docs/tags-and-labels#bucket-labels), expand the **Labels** section (), click **Add label** , and specify a `key` and a `value` for your label. 2. In the **Choose where to store your data** section, do the following: 1. Select a [Location type](https://docs.cloud.google.com/storage/docs/locations). 2. Choose a location where your bucket's data is permanently stored from the **[Location type](https://docs.cloud.google.com/storage/docs/locations#available-locations)** drop-down menu. - If you select the [dual-region](https://docs.cloud.google.com/storage/docs/locations#location-dr) location type, you can also choose to enable [turbo replication](https://docs.cloud.google.com/storage/docs/availability-durability#turbo-replication) by using the relevant checkbox. 3. To set up [cross-bucket replication](https://docs.cloud.google.com/storage/docs/availability-durability#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. Select a [default storage class](https://docs.cloud.google.com/storage/docs/storage-classes) for the bucket or [Autoclass](https://docs.cloud.google.com/storage/docs/autoclass) for automatic storage class management of your bucket's data. 2. To enable [hierarchical namespace](https://docs.cloud.google.com/storage/docs/hns-overview), in the **Optimize storage for data-intensive workloads** section, select **Enable hierarchical namespace on this bucket** . **Note:** You cannot enable hierarchical namespace in existing buckets. 4. In the **Choose how to control access to objects** section, select whether or not your bucket enforces [public access prevention](https://docs.cloud.google.com/storage/docs/public-access-prevention), and select an [access control method](https://docs.cloud.google.com/storage/docs/access-control) for your bucket's objects. **Note:** You cannot change the **Prevent public access** setting if this setting is enforced at an [organization policy](https://docs.cloud.google.com/storage/docs/org-policy-constraints#public-access-prevention). 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](https://docs.cloud.google.com/storage/docs/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](https://docs.cloud.google.com/storage/docs/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](https://docs.cloud.google.com/storage/docs/object-lock), click the **Enable object retention** checkbox. - To enable [Bucket Lock](https://docs.cloud.google.com/storage/docs/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](https://docs.cloud.google.com/storage/docs/encryption). 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](https://docs.cloud.google.com/compute/docs/regions-zones#available) of an existing or new Managed Service for Apache Spark 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 Managed Service for Apache Spark cluster.** Run the command, below, to create a [single-node](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/single-node-clusters) Managed Service for Apache Spark cluster in the specified [Compute Engine zone](https://docs.cloud.google.com/compute/docs/regions-zones). ``` gcloud dataproc clusters create ${CLUSTER} \     --project=${PROJECT} \     --region=${REGION} \     --single-node ``` > [!IMPORTANT] > The above command installs the default [cluster image version](https://docs.cloud.google.com/managed-spark/docs/concepts/versioning/dataproc-versions#supported_cloud_dataproc_versions). You can use the [`--image-version`](https://docs.cloud.google.com/sdk/gcloud/reference/dataproc/clusters/create#--image) 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](https://docs.cloud.google.com/managed-spark/docs/tutorials/gcs-connector-spark-tutorial#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: ``` gcloud storage cp gs://pub/shakespeare/rose.txt \     gs://${BUCKET_NAME}/input/rose.txt ```
6. **Set up a [Java (Apache Maven)](https://docs.cloud.google.com/java/docs/setup#install_a_build_automation_tool), [Scala (SBT)](https://www.scala-sbt.org/), or [Python](https://docs.cloud.google.com/python/docs/setup#installing_python) development environment.** > [!IMPORTANT] > **Use Cloud Shell** . [Cloud Shell](https://docs.cloud.google.com/shell/docs/features#tools) includes tools used in this tutorial, including Apache Maven, Python, and the Google Cloud CLI.
## 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](https://spark.apache.org/)** using [Apache Maven](https://maven.apache.org/) to build a [JAR](https://en.wikipedia.org/wiki/JAR) package - **[Spark job in Scala](https://www.scala-lang.org/)** using [SBT](https://www.scala-sbt.org/) to build a [JAR](https://en.wikipedia.org/wiki/JAR) package - **[Spark job in Python (PySpark)](https://archive.apache.org/dist/spark/docs/3.5.4/api/python/getting_started/index.html)** ### 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 Managed Service for Apache Spark 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 > [!IMPORTANT] > **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](https://docs.cloud.google.com/managed-spark/docs/concepts/versioning/dataproc-versions). Note that the `spark-core_` artifact number is the Scala `major.minor` version number. 4.0.0 dataproc.codelab word-count 1.0 1.8 1.8 org.scala-lang scala-library Scala version, for example, 2.11.8 provided org.apache.spark spark-core_Scala major.minor.version, for example, 2.11 Spark version, for example, 2.3.1 provided 2. **Copy the `WordCount.java` code listed, below, to your local machine.** 1. Create a set of directories with the path `src/main/java/managed-spark/codelab`: ``` mkdir -p src/main/java/managed-spark/codelab ``` 2. Copy `WordCount.java` to your local machine into `src/main/java/managed-spark/codelab`: ``` cp WordCount.java src/main/java/managed-spark/codelab ``` `WordCount.java` is a Spark job in Java that reads text files from Cloud Storage, performs a word count, then writes the text file results to Cloud Storage. ```java 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: "); } String inputPath = args[0]; String outputPath = args[1]; JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("Word Count")); JavaRDD lines = sparkContext.textFile(inputPath); JavaRDD words = lines.flatMap( (String line) -> Arrays.asList(line.split(" ")).iterator() ); JavaPairRDD 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/word-count-1.0.jar` is created. 4. **Stage the package to Cloud Storage.** ``` gcloud storage 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 Managed Service for Apache Spark 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 > [!IMPORTANT] > **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](https://docs.cloud.google.com/managed-spark/docs/concepts/versioning/dataproc-versions). ``` 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 Spark job in Java that reads text files from Cloud Storage, performs a word count, then writes the text file results to Cloud Storage. ```java 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: ") } 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.** ``` 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 1. **Copy `word-count.py` to your local machine.** This is a 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. ```java #!/usr/bin/env python import pyspark import sys if len(sys.argv) != 3: raise Exception("Exactly 2 arguments are required: ") 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 Managed Service for Apache Spark 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 gcloud CLI command to view the wordcount output. ``` gcloud storage 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.
### Delete the project The easiest way to eliminate billing is to delete the project that you created for the tutorial. To delete the project: > [!CAUTION] > **Caution** : Deleting a project has the following effects: > > - **Everything in the project is deleted.** If you used an existing project for the tasks in this document, 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 architectures, tutorials, or quickstarts, reusing projects > can help you avoid exceeding project quota limits. 1. In the Google Cloud console, go to the **Manage resources** page. [Go to Manage resources](https://console.cloud.google.com/iam-admin/projects) 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.

### Delete the Managed Service for Apache Spark cluster Instead of deleting your project, you might want to only [delete your cluster](https://docs.cloud.google.com/managed-spark/docs/guides/manage-cluster#deleting_a_cluster) within the project. ### Delete the Cloud Storage bucket ### Google Cloud console 1. In the Google Cloud console, go to the Cloud Storage **Buckets** page. [Go to Buckets](https://console.cloud.google.com/storage/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: ```bash gcloud storage buckets delete BUCKET_NAME ``` > [!NOTE] > **Important:** Your bucket must be empty before you can delete it. ## What's next - See [Spark job tuning tips](https://docs.cloud.google.com/managed-spark/docs/support/spark-job-tuning)