Spark Scala Job の提出

VPC環境で利用できます。

本ガイドでは、Spark Scala Jobを作成した後、Cloud Hadoopクラスタに提出する方法について説明します。

Scalaコードの作成とコンパイル

Scalaで Sparkアプリケーションを作成し、jarファイルとしてパッケージ化する方法には、以下のような2つの方法があります。

1. ターミナルで Scalaを使用する
2. IntelliJ SBTプラグインを使用する

1.ターミナルで Scalaを使用する

ターミナルで HelloScalaを出力する Scalaコードを作成し、コンパイルして jarファイルにパッケージ化する例を挙げて説明します。

Scalaバイナリファイルのダウンロード

Scalaバイナリをダウンロードして圧縮を解凍します。

MacOSで Homebrewを使用している場合は、以下のようにインストールできます。

brew install scala

環境変数設定

次のコマンドを使用して、実行ファイル(例: .bashrc)に SCALA_HOME 環境変数を設定し、PATHに $SCALA_HOMEを追加します。

export SCALA_HOME=/usr/local/share/scala
export PATH=$PATH:$SCALA_HOME/

Sparkアプリケーション作成

Sparkアプリケーションを作成する方法は、次の通りです。

1. scala REPLを実行して、scalaを実行します。

❯ scala
# Welcome to Scala version ...
# Type in expressions to have them evaluated.
# Type :help for more information.
# scala>

2. 次のように HelloWorld.scala classを作成して保存します。

object HelloWorld {
  def main(args: Array[String]): Unit = {
      println("Hello, world!")
  }
}

scala> :save HelloWorld.scala
scala> :q

3. 以下のコマンドを使用して scalacにコンパイルします。

❯ scalac  HelloWorld.scala

4. ls コマンドを使用して、正常にコンパイルされたかを.classファイルで確認します。

❯ ls HelloWorld*.class
HelloWorld$.class HelloWorld.class

jarファイル作成

jarファイルを作成する方法は、次の通りです。

1. HelloWorld*.classファイルがあるディレクトリに移動した後、次のコマンドを使用してクラスファイルを jarファイルにパッケージ化します。

❯ jar cvfe HelloWorld.jar HelloWorld HelloWorld*.class
added manifest
adding: HelloWorld$.class(in = 670) (out= 432)(deflated 35%)
adding: HelloWorld.class(in = 645) (out= 524)(deflated 18%)

2. パッケージ化された jarファイル内で、HelloWorld classがアプリケーションの entry pointとして設定されているかを MANIFEST.MFで確認します。

❯ unzip -q -c HelloWorld.jar META-INF/MANIFEST.MF
Manifest-Version: 1.0
Created-By: 1.8.0_181 (Oracle Corporation)
Main-Class: HelloWorld # entry point

2.IntelliJ SBTプラグインを使用する

本ガイドでは、Sparkアプリケーションの開発およびデバッグを行うための環境を IntelliJに設定し、Hello Scalaという WordCount Jobを構築する方法を例に挙げて説明します。

• ビルドマネージャ: SBT
• 作成環境の例: Windows OS、IntelliJ Ultimate 2022.1.4

プロジェクト作成

プロジェクトを作成する方法は、次の通りです。

IntelliJを実行します。

1. 左側の Pluginsメニューから Scalaを検索してインストールします。
   

2. プラグインを反映させるには再起動が必要です。[Restart IDE] ボタンをクリックして、IntelliJを再起動します。
   

3. ホーム画面左側のメニューから Projectsをクリックし、New Projectをクリックします。
   

4. 次のように Scalaと sbtを選択し、 [Create] ボタンをクリックします。

   • プロジェクト名: WordCountに指定
   • Scalaと sbtのバージョンを選択
     

5. プロジェクトが正常に作成されたか確認します。

   • プロジェクトが作成されると、デフォルトでは次のような構造のディレクトリおよびファイル情報を確認できます。
     • .idea: IntelliJ構成ファイル
     • project: コンパイルに使用されるファイル
     • src: ソースコード。アプリケーションコードのほとんどは src/mainに配置する必要がある。src/testはテストスクリプトのためのスペース
     • target: プロジェクトをコンパイルするとこの場所に保存
     • build.sbt: SBT構成ファイル
       

SBTライブラリのインポート

IntelliJが Sparkコードを認識するには、spark-coreライブラリとドキュメントをインポートする必要があります。

1. mvn repositoryで、spark-coreライブラリと Artifact Idと併せて互換性のある Scalaのバージョンを確認します。

   

2. Target > build.sbtをクリックし、スクリプトウィンドウに以下の内容を追加します。

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"

3. ライブラリが正常にインポートされたか、Buildコンソールで確認します。
   

Group Id %% Artifact Id % Revision

Sparkアプリケーション作成

ここでは、シェイクスピアのソネットのテキストファイル(shakespeare.txt)をデータセットとして使用し、ソネットに含まれる単語を数える WordCountアプリケーションの作成方法を例に挙げて説明します。

1. shakespeare.txtをダウンロードして src/main/resourcesに保存します。

   • Cloud Hadoopクラスタでこのアプリケーションを実行する際は、S3バケットまたは HDFSにデータセットをアップロードして使用します。
     

2. src > mainを選択してディレクトリを拡張し、scalaディレクトリを右クリックして New > Scala Classをクリックします。

3. WordCount/src/main/scalaの下位に新しいクラスを作成します。

   • Kind: Object
     

4. 正しく設定されているか確認するため、WordCount.scalaに以下のサンプルコードを作成して実行します。

object WordCount {
    def main(args: Array[String]): Unit = {
      println("This is WordCount application")
    }
}

5. 結果が正常に出力されているか確認します。
   

6. WordCount.scalaに適用していたサンプルコードを削除した後、シェイクスピアのソネットのテキストファイルの単語数を数えるコードを以下のように作成します。

import org.apache.spark.{SparkConf, SparkContext}

object WordCount {

def main(args: Array[String]) : Unit = {

  //Create a SparkContext to initialize Spark
  val conf = new SparkConf()
  conf.setMaster("local")
  conf.setAppName("Word Count")
  val sc = new SparkContext(conf)

  // Load the text into a Spark RDD, which is a distributed representation of each line of text
  val textFile = sc.textFile("src/main/resources/shakespeare.txt")

  //word count
  val counts = textFile.flatMap(line => line.split(" "))
    .map(word => (word, 1))
    .reduceByKey(_ + _)

  counts.foreach(println)
  System.out.println("Total words: " + counts.count())
  counts.saveAsTextFile("/tmp/sonnetWordCount")
}

}

7. WordCount.scalaを実行して出力結果を確認します。
   

jarファイル作成

1. Object Storageのバケットにデータセットをアップロードした後、ソースコード内の resourceファイルパスを次のように変更します。
   • データセットを HDFSにアップロードして使用する場合は、s3a:// の代わりに hdfs://を使用します。

// FROM
conf.setMaster("local")
// TO
conf.setMaster("yarn-cluster")

// FROM
val textFile = sc.textFile("src/main/resources/shakespeare.txt")
// TO
val textFile = sc.textFile("s3a://deepdrive-hue/tmp/shakespeare.txt")

// FROM
counts.saveAsTextFile("/tmp/sonnetWordCount");
// TO
counts.saveAsTextFile("s3a://deepdrive-hue/tmp/sonnetWordCount");

2. ターミナルコンソールで次のコマンドを使用して、アップデートしたコードを Cloud Hadoopクラスタに提出できるよう、compiled jarとしてパッケージ化します。
   • jarファイルには、アプリケーションコードと build.sbtで定義されたすべての dependenciesが含まれています。
   • sbt package コマンドは $PROJECT_HOME/target/scala-2.11 の下位に wordcount_2.11-0.1.jarファイルを作成します。

> cd ~/IdeaProjects/WordCount # PROJECT_HOME
> sbt package

Cloud Hadoopクラスタへの Spark Job提出

ローカルで作成した Sparkアプリケーション(.jar)を Cloud Hadoopにリリースして実行する方法について説明します。

Object Storageへの jarsアップロード

Hueの S3ブラウザまたは Object Storageコンソールを使用して、shakespeare.txtと jarを Object Storageバケットにコピーします。

• Hueへのアクセスおよび使用方法の詳細は、Hue を使用するガイドをご参照ください。
• Object Storageバケットに関する詳細は、Object Storage の概要ガイドをご参照ください。
  

Job提出

jarファイルをクラスタに提出する2つの方法について説明します。

spark.hadoop.fs.s3a.access.key <OBJECT-STORAGE-ACCESS-KEY>
spark.hadoop.fs.s3a.endpoint   kr.objectstorage.ncloud.com
spark.hadoop.fs.s3a.secret.key <OBJECT-STORAGE-SECRET-KEY>

• Hueでの Spark Submit Jar利用
  

• Sparkクライアントノードからの提出

1. Sparkクライアントがインストールされているクラスタノードで、次のように spark-submit コマンドを実行します。

spark-submit --class WordCount --master yarn-cluster --deploy-mode cluster s3a://deepdrive-hue/tmp/wordcount_2.11-0.1.jar

2. Jobの実行が完了したら、以下のように結果が指定したバケットパス内に保存されているか確認します。