「Managed Service for Apache Spark」は、以前は「Compute Engine 上の Dataproc」（クラスタデプロイ）と「Apache Spark 用 Google Cloud Serverless」（サーバーレスデプロイ）と呼ばれていたプロダクトの新しい名前です。

Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Cloud Storage コネクタと Apache Spark の併用

このチュートリアルでは、 Cloud Storage コネクタと Apache Sparkを併用するサンプルコードの実行方法について説明します。

Lightning Engine は Cloud Storage への接続を強化し、ネイティブエンジンのパフォーマンスを最適化します。改良された Cloud Storage コネクタにより、メタデータオペレーションを最小限に抑えることで費用の削減が実現し、最適化されたファイル出力コミッターにより、Spark ワークロードのパフォーマンスと信頼性が高まります。この限定公開プレビュー機能への早期アクセスをリクエストするには、早期アクセスフォームに記入してください。

目標

Java、Scala、または Python で簡単な wordcount Spark ジョブを作成し、Managed Service for Apache Spark クラスタでジョブを実行します。

費用

このドキュメントでは、課金対象である次のコンポーネントを使用します Google Cloud：

Compute Engine
Managed Service for Apache Spark
Cloud Storage

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを生成できます。

新規の Google Cloud ユーザーは無料トライアルをご利用いただける場合があります。

始める前に

以下の手順を実行して、このチュートリアルでコードを実行する準備をします。

プロジェクトを設定します。必要に応じて、Managed Service for Apache Spark、Compute Engine、Cloud Storage API を有効にし、ローカルマシンに Google Cloud CLI をインストールしてプロジェクトを設定します。
Cloud Storage バケットを作成します。Cloud Storage はチュートリアルデータを保持するために必要です。使用できるバケットがない場合は、プロジェクトに新しいバケットを作成します。
1. コンソールで Cloud Storage の Google Cloud [**バケット**] ページに移動します。
  [バケット] に移動
2. [ [Create]] をクリックします。
3. [バケットの作成] ページでユーザーのバケット情報を入力します。次のステップに進むには、[続行] をクリックします。
  1. [スタートガイド] セクションで、次の操作を行います。
    - バケット名の要件を満たす、グローバルに一意の名前を入力します。
    - バケットラベルを追加するには、[ラベル] セクション（）を開き、[ラベルを追加] をクリックして、key と value を指定します。
  2. [データの保存場所の選択] セクションで、次の操作を行います。
    1. ロケーションタイプを選択してください。
    2. [Location type] プルダウンメニューから、バケットのデータが永続的に保存されるロケーションを選択します。
      - ロケーションタイプとして [デュアルリージョン] を選択した場合は、関連するチェックボックスを使用して [ターボレプリケーション] を有効にすることもできます。
    3. クロスバケットレプリケーションを設定するには、 [Storage Transfer Service 経由でクロスバケットレプリケーションを追加する] を選択し、次の手順を実施します:
      クロスバケットレプリケーションを設定する
      
      [バケット] メニューで、バケットを選択します。
      
      [レプリケーション設定] セクションで、[構成] をクリックして、レプリケーションジョブの設定を構成します。
      
      [**クロスバケットレプリケーションを構成する**] ペインが表示されます。
      
      オブジェクト名の接頭辞で複製するオブジェクトをフィルタするには、オブジェクトを追加または除外する接頭辞を入力し、 [接頭辞を追加] をクリックします。
      
      複製されたオブジェクトのストレージクラスを設定するには、 [Storage class] メニューからストレージクラスを選択します。この手順をスキップすると、複製されたオブジェクトはデフォルトで宛先バケットのストレージクラスを使用します。
      
      [完了] をクリックします。
  3. [データの保存場所の選択] セクションで、次の操作を行います。
    1. バケットのデフォルトのストレージクラスを選択するか、バケットデータのストレージクラスを自動的に管理するAutoclassを選択します。
    2. 階層名前空間を有効にするには、 [データ量が多いワークロード向けにストレージを最適化] セクションで、 [このバケットで階層的な名前空間を有効にする] を選択します。
      注: 既存のバケットで階層名前空間を有効にすることはできません。
  4. In the [オブジェクトへのアクセスを制御する方法を選択する] セクションで、バケットに公開アクセスの防止を適用するかどうかを選択し、バケットのオブジェクトに使用するアクセス制御方法を選択します。
    **注:** 組織のポリシーでこの設定が適用されている場合、[**公開アクセスの防止**] 設定を変更することはできません。
  5. [オブジェクトデータを保護する方法を選択する] セクションで、次の操作を行います。
    - [**データ保護**] で、バケットに設定するオプションを選択します。
      - 削除（復元可能）を有効にするには、 [削除（復元可能）ポリシー（データ復旧用）] チェックボックスをオンにして、削除後にオブジェクトを保持する日数を指定します。
      - オブジェクトのバージョニングを設定するには、 [オブジェクトのバージョニング（バージョン管理用）] チェックボックスをオンにして、オブジェクトごとの最大バージョン数と、非現行バージョンが期限切れになるまでの日数を指定します。
      - オブジェクトとバケットの保持ポリシーを有効にするには、[保持（コンプライアンス用）] チェックボックスをオンにして、次の操作を行います。
        
        [オブジェクト保持ロック]を有効にするには、 [オブジェクト保持を有効にする]チェックボックスをオンにします。
        
        [Bucket Lock] を有効にするには、[バケット保持ポリシーを設定する] チェックボックスをオンにして、保持期間の単位と保持期間を選択します。
    - オブジェクトデータの暗号化方法を選択するには、 [データ暗号化] セクション（）を開き、 [データの暗号化] 方法を選択します。
4. [作成] をクリックします。
ローカル環境変数を設定します。ローカルマシンで環境変数を設定します。プロジェクト ID と、このチュートリアルで使用する Cloud Storage バケットの名前を設定します。 Google Cloud 既存または新規の Managed Service for Apache Spark クラスタの名前とリージョンも指定します。次の手順で、このチュートリアルで使用するクラスタを作成できます。
```
PROJECT=project-id
```
```
BUCKET_NAME=bucket-name
```
```
CLUSTER=cluster-name
```
```
REGION=cluster-region Example: "us-central1"
```
Managed Service for Apache Spark クラスタを作成します。以下のコマンドを実行して、指定された Compute Engine ゾーンに単一ノードの Managed Service for Apache Spark クラスタを作成します。
```
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 ジョブ。JAR パッケージを構築する際に Apache Maven を使用
Scala の Spark ジョブ。JAR パッケージを構築する際に SBT を使用
Python の Spark ジョブ（PySpark）

Java

pom.xml ファイルをローカルマシンにコピーします。 次の pom.xml ファイルで、Scala ライブラリと Spark ライブラリの依存関係を指定します。この依存関係には、実行時に Managed Service for Apache Spark クラスタがこれらのライブラリを提供することを示す provided スコープが指定されます。pom.xml ファイルでは、コネクタが標準の HDFS インターフェースを実装しているため、Cloud Storage の依存関係は指定されません。Spark ジョブが Cloud Storage クラスタファイル（gs:// で始まる URI を持つファイル）にアクセスすると、システムは自動的に 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 ライブラリの依存関係を指定します。この依存関係には、実行時に Managed Service for Apache Spark クラスタがこれらのライブラリを提供することを示す provided スコープが指定されます。build.sbt ファイルでは、コネクタが標準の HDFS インターフェースを実装しているため、Cloud Storage の依存関係は指定されません。Spark ジョブが Cloud Storage クラスタファイル（gs:// で始まる URI を持つファイル）にアクセスすると、システムは自動的に 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 をコピーします。これは PySpark を使用した Python の 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 ジョブを Managed Service for Apache Spark クラスタに送信します。

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 コマンドを実行して wordcount の出力を表示します。

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

wordcount の出力は、次のようになります。

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

クリーンアップ

チュートリアルが終了したら、作成したリソースをクリーンアップして、割り当ての使用を停止し、課金されないようにできます。次のセクションで、リソースを削除または無効にする方法を説明します。

プロジェクトの削除

課金されないようにする最も簡単な方法は、チュートリアル用に作成したプロジェクトを削除することです。

プロジェクトを削除するには:

注意 : プロジェクトを削除すると、次のような影響があります。

プロジェクト内のすべてのものが削除されます。既存のプロジェクトを使用してこのドキュメントのタスクを行った場合、そのプロジェクトを削除すると、プロジェクト内で行った他の作業もすべて削除されます。
カスタムプロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタムプロジェクト ID を作成した可能性があります。そのプロジェクト ID を使用した URL（例: appspot.com）を保持するには、プロジェクト全体ではなく、プロジェクト内の選択したリソースだけを削除します。

複数のアーキテクチャ、チュートリアル、クイックスタートを実施する予定がある場合は、プロジェクトを再利用するとプロジェクトの割り当て上限を超えないようにすることができます。

コンソールで [**リソースの管理**] ページに移動します。 Google Cloud
[リソースの管理] に移動
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、 [Shut down] をクリックしてプロジェクトを削除します。

Managed Service for Apache Spark クラスタを削除する

プロジェクトを削除する代わりに、プロジェクト内のクラスタのみを削除することもできます。

Cloud Storage バケットを削除する

Google Cloud コンソール

コンソールで Cloud Storage の [**バケット**] ページに移動します。 Google Cloud
[バケット] に移動
削除するバケットのチェックボックスをクリックします。
バケットを削除するには、 [Delete] をクリックして、指示に沿って操作します。

コマンドライン

gcloud storage buckets delete BUCKET_NAME

次のステップ

Spark ジョブ調整のヒントを見る。

Cloud Storage コネクタと Apache Spark の併用 コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

目標

費用

始める前に

クロスバケット レプリケーションを設定する

Spark wordcount ジョブを準備する

Java

Scala

Python

ジョブを送信する

Java

Scala

Python

出力を表示する

クリーンアップ

プロジェクトの削除

Managed Service for Apache Spark クラスタを削除する

Cloud Storage バケットを削除する

Google Cloud コンソール

コマンドライン

次のステップ

Cloud Storage コネクタと Apache Spark の併用

クロスバケットレプリケーションを設定する