SparkからS3のデータを読み込みんでみる

Apache Spark AWS

AWSでの準備

１． AWSマネージメント コンソールにアクセス。
２． IAMに移動。
３． 「ユーザー」でユーザーを作成。
アクセスキーをメモ。
４． 作成したユーザーを選んで、
「ポリシーのアタッチ」
AmazonS3ReadOnlyAccess
を選んで、アタッチする。

IntelliJの設定

前に試したSparkのプロジェクト、
Spark 1.5.0の多層パーセプトロンのサンプルコードを試す(Mac) - kubotti’s memo
に新しくscalaファイルを作って実行したら動いた。
src/main/scala/S3Access.scala

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

object S3Access {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("S3 Sample Application").setMaster("local")
    val sc = new SparkContext(conf)
    val hadoopConf=sc.hadoopConfiguration;
    hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
    //hadoopConf.set("fs.s3.awsAccessKeyId",myAccessKey)
    //hadoopConf.set("fs.s3.awsSecretAccessKey",mySecretKey)

    //val data = sc.textFile("s3n://(bucket)/(path))
    val bucket = "bucketname"
    val filename = "s3filename"
    val data = sc.textFile("s3n://" + bucket + "/" + filename)
    val numHttp = data.filter(line => line.contains("HTTP")).count()
    val numIndex = data.filter(line => line.contains("index.html")).count()
    println("numHttp: %s".format(numHttp))
    println("numIndex: %s".format(numIndex))
    System.exit(0)
  }
}

Run Configuration の Environmental Variables（環境変数）に
AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY
を設定。

↓この記事を参考にしたけど、AWSのアクセスキーは環境変数に設定すればソースコードには書かなくても動くっぽい。
http://itsneatlife.blogspot.jp/2014/01/sparks3.html

アクセスキーが間違っていた場合

15/10/02 16:36:04 INFO SparkContext: Created broadcast 0 from textFile at S3Access.scala:16
Exception in thread "main" org.apache.hadoop.fs.s3.S3Exception: org.jets3t.service.S3ServiceException: S3 HEAD request failed for '/2015-07-30-05-23-20-3340D0370B9F3B0A' - ResponseCode=403, ResponseMessage=Forbidden
    at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.handleServiceException(Jets3tNativeFileSystemStore.java:245)
    at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:119)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:186)

S3の複数ファイル

あと、検索したときに出てきた記事によると、
s3の大量のファイルをワイルドカードで読み込むと、問題が起きるらしい。
あとで読む。
sc.textFile()を使うのはよくないと書かれてる（気がする）。
How NOT to pull from S3 using Apache Spark