Zenn
🐇

ゼロから AWS Glue に Spark ジョブ作成

2021/07/31に公開
AWS
Scala
Glue
#
spark
tech

手元で sbt プロジェクトを作成するところから aws glue に spark ジョブ(Scala) を作成するまでの手順です。

Apache Spark インストール

brew install apache-spark

SBTプロジェクトのセットアップ

往年のコマンドでSBTプロジェクトを作成しましょう。
今回、プロジェクト名は hello-glue-scala-spark とします。

プロジェクト作成

sbt new sbt/scala-seed.g8
...
name [Scala Seed Project]: hello-glue-scala-spark
...

サンプルコードは消しちゃいます。

rm -rf src/test/scala/example

sbt-assembly を準備

jar ファイルを作りたいので sbt-assembly を使えるようにしておきます。

plugin.sbt
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "1.0.0")

build.sbt の編集

build.sbt を下記のように編集します。

build.sbt
import Dependencies._

lazy val root = (project in file("."))
  .settings(
    name := "hello-glue-scala-spark",
    scalaVersion := "2.11.12",
    assembly / assemblyJarName := "app.jar",
    assembly / assemblyOption := (assembly / assemblyOption).value.copy(cacheOutput = false),
    assembly / assemblyMergeStrategy := {
      case PathList("META-INF", xs @ _*) => MergeStrategy.discard
      case x                             => MergeStrategy.first
    },
    assembly / assemblyShadeRules := Seq(
      ShadeRule.rename("org.apache.http.**" -> "org.apache.httpShaded").inAll
    ),
    resolvers ++= Seq(
      "aws-glue-etl-artifacts" at "https://aws-glue-etl-artifacts.s3.amazonaws.com/release/"
    ),
    libraryDependencies ++= Seq(
      "com.amazonaws"          % "AWSGlueETL" % "1.0.0"   % "provided",
      "org.apache.spark"      %% "spark-core" % "2.4.3"   % "provided",
      "org.apache.spark"      %% "spark-sql"  % "2.4.3"   % "provided"
    )
  )

スクリプトの実装

src/main/scala/HelloSparkApp.scala を作成します。

オブジェクト名 説明
HelloSparkApp Glue で JOB を実行する際のメインモジュール
HelloSparkAppOnLocal ローカル実行用のモジュール
StdoutPeople Spark のロジックを切り出したオブジェクト
src/main/scala/HelloSparkApp.scala
import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.{GlueArgParser, Job}
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession

import scala.collection.JavaConverters._

object HelloSparkApp {

  def main(sysArgs: Array[String]) {
    val spark: SparkContext      = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)

    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("TempDir", "JOB_NAME", "Message").toArray)
    println(args("Message"))

    Job.init(args("JOB_NAME"), glueContext, args.asJava)

    StdoutPeople.run()

    Job.commit()
  }
}

object StdoutPeople {

  def run(): Unit = {

    val spark = SparkSession.builder.appName(this.getClass.getName).getOrCreate()
    import spark.implicits._

    val people = Seq(
      ("Spark太郎", "spark-taro@example.com", "2021-01-01", 11),
      ("Spark二郎", "spark-jiro@example.com", "2021-02-02", 12),
      ("Spark三郎", "spark-saburo@example.com", "2021-03-03", 13)
    ).toDF("name", "email", "birthday", "age")

    people.printSchema()
    people.show(3)
  }

}

object HelloSparkAppOnLocal {
  def main(args: Array[String]): Unit = StdoutPeople.run()
}

手元での動作確認

下記の手順を実施してスクリプトを動かしてみます。

  • sbt assembly を実行して jar ファイルを生成
  • spark-submit でローカルモジュール実行
sbt assembly

spark-submit --class HelloSparkAppOnLocal target/scala-2.11/app.jar

...

root
 |-- name: string (nullable = true)
 |-- email: string (nullable = true)
 |-- birthday: string (nullable = true)
 |-- age: integer (nullable = false)
 
 ...
 
+---------+--------------------+----------+---+
|     name|               email|  birthday|age|
+---------+--------------------+----------+---+
|Spark太郎|spark-taro@exampl...|2021-01-01| 11|
|Spark二郎|spark-jiro@exampl...|2021-02-02| 12|
|Spark三郎|spark-saburo@exam...|2021-03-03| 13|
+---------+--------------------+----------+---+

Glueへジョブの定義を作成

実装した Script と jar ファイルをS3へアップロード

jar ファイルだけかと思いきや、HelloSparkApp.Scala もアップロードする必要があるので注意してください。

export BUCKET=<your-deployment-bucket>
aws s3 cp target/scala-2.11/app.jar s3://${BUCKET}/app.jar
aws s3 sync src/main/scala/ s3://${BUCKET}/

バケット直下が下記のような状態になっていればOKです。

IAM Role 作成

ジョブに指定するIAM ロールを作成します。

  • IAM ロールを作成します。作成したロールを Glue が Assume Role できるように設定しておきます。
  • 作成したロールに Glue のマネージドポリシーをアタッチします。
cat glue-trust.json
{
    "Version": "2012-10-17",
    "Statement": {
        "Effect": "Allow",
        "Principal": {"Service": "glue.amazonaws.com"},
        "Action": "sts:AssumeRole"
    }
}


export ROLE_NAME=hello-glue-scala-spark-role
aws iam create-role \
	--role-name ${ROLE_NAME} \
	--assume-role-policy-document file://glue-trust.json
    
aws iam attach-role-policy \
	--role-name ${ROLE_NAME} \
	--policy-arn arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole

Glue のマネージドポリシーがアタッチされていることを確認します。

信頼されたエンティティに glue.amazonaws.com が指定されていることを確認します。

ジョブ作成

S3へアップロードしたスクリプト、jar ファイルのオブジェクトパス、IAMロールを指定してGlueにジョブの定義を追加します。

export REGION=ap-northeast-1
aws glue create-job \
     --name hello-spark \
     --role ${ROLE_NAME} \
     --glue-version 2.0 \
     --command "{ \
       \"Name\": \"glueetl\",\
       \"ScriptLocation\": \"s3://${BUCKET}/HelloSparkApp.scala\" \
     }" \
     --region ${REGION} \
     --output json \
     --default-arguments "{ \
       \"--job-language\": \"scala\", \
       \"--class\": \"HelloSparkApp\", \
       \"--extra-jars\": \"s3://${BUCKET}/app.jar\", \
       \"--Message\": \"HELLO SPARK!!!\" \
     }" \
     --endpoint https://glue.${REGION}.amazonaws.com

作成したジョブの定義を確認します。

動かしてみる

さっそくジョブを動かしてみます。コマンドを叩くと JobRunId が返ってきます。

aws glue start-job-run --job-name hello-spark
{
    "JobRunId": "jr_3e43521ccbfe86cac1689200b929c61c74c5b5cee3c4e6663b706a19e753f1cf"
}

履歴 のタブから実行ステータスが確認できます。

終わりました。成功したみたいです。

ログを確認してみましょう。CloudWatch に 出力されています。

Glue のジョブを作成して動かすことができました。

ハマったところ

  • build.sbt の libraryDependencies で AWSGlueETL を追加する際に provided の指定を忘れていたら sbt assembly が15分くらいかかってしまい頭を悩ませてました。
    • 詳しい人に聞いてみたら Spark の依存を詰め込もうとして重くなっていたのではとのことでした。
  • aws glue create-job でうっかり ScriptLocation に jar ファイルのオブジェクトパスを指定して、当然動くハズもなくハマりました。
    • ScriptLocation へは .scala ファイルを、--default-arguments の --extra-jars へ jar ファイルのパスを追加しなければならないということでした。

とても苦戦しました。

おわり

サンプルコードはこちら takat0-h0rikosh1/aws-glue-scala-spark-template

Discussion