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

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

Spark wordcount ジョブを準備する

以下のタブを選択し、手順に沿ってクラスタに送信するジョブ パッケージまたはファイルを準備します。次のいずれかのジョブタイプを準備できます。

• Java の Spark ジョブ。JAR パッケージのビルドに Apache Maven を使用
• Scala の Spark ジョブ。JAR パッケージのビルドに SBT を使用
• Python の Spark ジョブ（PySpark）

Java

1. pom.xml ファイルをローカルマシンにコピーします。次の pom.xml ファイルで、Scala ライブラリと Spark ライブラリの依存関係を指定します。この依存関係には、実行時に Dataproc クラスタがこれらのライブラリを提供することを示す 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>

2. 以下の WordCount.java コードをローカルマシンにコピーします。
   1. パス src/main/java/dataproc/codelab を指定して一連のディレクトリを作成します。

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

   2. 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);
     }
   }

3. パッケージをビルドします。

   mvn clean package
   

   ビルドが成功すると、target/word-count-1.0.jar が作成されます。
4. パッケージを Cloud Storage にステージングします。

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

Scala

1. build.sbt ファイルをローカルマシンにコピーします。次の build.sbt ファイルで、Scala ライブラリと Spark ライブラリの依存関係を指定します。この依存関係には、実行時に Dataproc クラスタがこれらのライブラリを提供することを示す 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"
   )

2. ローカルマシンに 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)
     }
   }

3. パッケージをビルドします。

   sbt clean package
   

   ビルドが成功すると、target/scala-2.11/word-count_2.11-1.0.jar が作成されます。
4. パッケージを 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. ローカルマシンに 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 ジョブを 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 コマンドを実行して 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)