此页面由 Cloud Translation API 翻译。

将 Cloud Storage 连接器与 Apache Spark 搭配使用

本教程介绍了如何执行将 Cloud Storage 连接器与 Apache Spark 搭配使用的示例代码。

目标

使用 Java、Scala 或 Python 编写简单的 wordcount Spark 作业，然后在 Dataproc 集群上运行该作业。

费用

在本文档中，您将使用 Google Cloud 的以下收费组件：

Compute Engine
Dataproc
Cloud Storage

您可使用价格计算器根据您的预计使用情况来估算费用。 Google Cloud 新用户可能有资格申请免费试用。

准备工作

运行以下步骤为运行本教程中的代码做准备。

设置项目。如有必要，请设置一个启用了 Dataproc、Compute Engine 和 Cloud Storage API 并在本地机器上安装了 Google Cloud CLI 的项目。
创建 Cloud Storage 存储分区。您需要使用 Cloud Storage 来保存教程数据。如果您没有可用的 Cloud Storage 存储分区，请在项目中创建新存储分区。
1. In the Google Cloud console, go to the Cloud Storage Buckets page.
  Go to Buckets page
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.
设置本地环境变量。在本地机器上设置环境变量。设置 Google Cloud 项目 ID 以及您将在本教程中使用的 Cloud Storage 存储分区的名称。此外，还要提供现有或新 Dataproc 集群的名称和区域。您可以在下一步中创建要用于本教程的集群。
```
PROJECT=project-id
```
```
BUCKET_NAME=bucket-name
```
```
CLUSTER=cluster-name
```
```
REGION=cluster-region Example: "us-central1"
```
创建 Dataproc 集群。运行以下命令，在指定的 Compute Engine 地区中创建单节点 Dataproc 集群。
```
gcloud dataproc clusters create ${CLUSTER} \
    --project=${PROJECT} \
    --region=${REGION} \
    --single-node
```
以上命令用于安装默认的集群映像版本。您可以使用 --image-version 标志为集群选择映像版本。每个映像版本都会安装特定版本的 Spark 和 Scala 库组件。如果您准备 Java 或 Scala 中的 Spark wordcount 作业，则在准备作业软件包时将引用集群上安装的 Spark 和 Scala 版本。
将公共数据复制到 Cloud Storage 存储分区。将公共数据“莎士比亚”文本片段复制到 Cloud Storage 存储桶的 input 文件夹中：
```
gcloud storage cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
设置 Java (Apache Maven)、Scala (SBT) 或 Python 开发环境。
使用 Cloud Shell。 Cloud Shell 包括本教程中使用的工具，包括 Apache Maven、Python 和 Google Cloud CLI。

准备 Spark wordcount 作业

在下面选择一个标签页，按照步骤准备作业软件包或文件以提交到集群。您可以准备以下作业类型之一：

Java 中的 Spark 作业使用 Apache Maven 创建一个 JAR 软件包
Scala 中的 Spark 作业使用 SBT 创建一个 JAR 软件包
Python (PySpark) 中的 Spark 作业

Java

将 pom.xml 文件复制到本地机器。以下 pom.xml 文件指定 Scala 和 Spark 库依赖项，它们具有 provided 范围，指示 Dataproc 集群将在运行时提供这些库。pom.xml 文件未指定 Cloud Storage 依赖项，因为连接器实现了标准 HDFS 接口。当 Spark 作业访问 Cloud Storage 集群文件时（URI 开头为 gs:// 的文件），系统会自动使用 Cloud Storage 连接器访问 Cloud Storage 中的文件

检查您的集群映像版本。替换文件中的 version 占位符，以显示集群的映像版本使用的 Spark 和 Scala 库版本。请注意，spark-core_ 工件数是 Scala major.minor 版本号。

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

将下面列出的 WordCount.java 代码复制到您的本地机器。

创建一组路径为 src/main/java/dataproc/codelab 的目录：
```
mkdir -p src/main/java/dataproc/codelab
```
将 WordCount.java 复制到您的本地机器的 src/main/java/dataproc/codelab：
```
cp WordCount.java src/main/java/dataproc/codelab
```

WordCount.java 是 Java 中一个简单的 Spark 作业，可从 Cloud Storage 读取文本文件、执行字数统计，然后将文本文件结果写入 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);
  }
}

构建软件包。
```
mvn clean package
```
如果构建成功，则系统会创建一个 target/word-count-1.0.jar。

将软件包暂存到 Cloud Storage。

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

Scala

将 build.sbt 文件复制到本地机器。以下 build.sbt 文件指定 Scala 和 Spark 库依赖项，它们具有 provided 范围，指示 Dataproc 集群将在运行时提供这些库。build.sbt 文件未指定 Cloud Storage 依赖项，因为连接器实现了标准 HDFS 接口。当 Spark 作业访问 Cloud Storage 集群文件时（URI 开头为 gs:// 的文件），系统会自动使用 Cloud Storage 连接器访问 Cloud Storage 中的文件
检查您的集群映像版本。替换文件中的 version 占位符，以显示集群的映像版本使用的 Spark 和 Scala 库版本。
```
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"
)
```

将 word-count.scala 复制到本地机器。这是 Java 中一个简单的 Spark 作业，可从 Cloud Storage 读取文本文件、执行字数统计，然后将文本文件结果写入 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)
  }
}

构建软件包。
```
sbt clean package
```
如果构建成功，则系统会创建一个 target/scala-2.11/word-count_2.11-1.0.jar。

将软件包暂存到 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

将 word-count.py 复制到本地机器。这是 Python 中一个简单的使用 PySpark 的 Spark 作业，可从 Cloud Storage 读取文本文件、执行字数统计，然后将文本文件结果写入 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])

提交作业

运行以下 gcloud 命令，将 Wordcount 作业提交到 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/

查看输出

作业完成后，运行以下 gcloud CLI 命令以查看字数统计输出。

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

字数统计输出应类似于以下内容：

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

清除数据

完成本教程后，您可以清理您创建的资源，让它们停止使用配额，以免产生费用。以下部分介绍如何删除或关闭这些资源。

删除项目

为了避免产生费用，最简单的方法是删除您为本教程创建的项目。

要删除项目，请执行以下操作：

In the Google Cloud console, go to the Manage resources page.
Go to Manage resources
In the project list, select the project that you want to delete, and then click Delete.
In the dialog, type the project ID, and then click Shut down to delete the project.

删除 Dataproc 集群

您可能希望只删除项目中的集群，而不是项目。

删除 Cloud Storage 存储分区

Google Cloud 控制台

In the Google Cloud console, go to the Cloud Storage Buckets page.
Go to Buckets
Click the checkbox for the bucket that you want to delete.
To delete the bucket, click Delete, and then follow the instructions.

命令行

gcloud storage buckets delete BUCKET_NAME

后续步骤

请参阅 Spark 作业微调提示