Lambdaが呼び出す場所

EMR+Lambdaでバッチ処理を自動化する（Lambdaの準備編）でLambdaが実行するコード指定に

パッケージ.クラス::メソッド

# 今回は, hoge.Hoge::run のようになります

があったんですが、実際の中身になります。

public class Hoge {

    /**
     * 実行
     * @param data
     * @param context
     * @return
     */
    public String run(final Map<String, String> data, final Context context) {
    
    	//EMR起動（後述）
    
    	"complete!"
    }

第１引数がMap<String, String>なのは、Jsonが来る想定になっています。Jsonがネストする場合、Map<String, Object>が正解かも。

Stringで受けてしまうとエラーが起こるっぽいので注意

EMRの起動について

EMRを使ったことがある方は既にお分かりの方が多いと思いますが、EMRを設定するためには

EMRが利用するソフトウェア(Hadoop, Sparkなど）の設定
EMRが立ち上げるEC2インスタンスの設定
それぞれのセキュリティの設定
EMRで何をするかのStepの設定

の４つ程度に分かれるとおもいます。
それぞれどのように設定していくか、紹介します。

EMRが利用するソフトウェア(Hadoop, Sparkなど）の設定

RunJobFlowRequest job = new RunJobFlowRequest();
job.withName("EMRの名前");

// 現在私は、emr-5.0.0を使いたいので "emr-5.0.0" と、入れています。
job.withReleaseLabel("EMRのバージョン");

job.withLogUri("ログを排出するS3のパス s3://~~~");
job.withServiceRole("EMRに設定するIamRole");
job.withJobFlowRole("EMRが起動するEC2に設定するIamRole");
job.withVisibleToAllUsers(true);
job.withInstances("EMRが立ち上げるEC2インスタンスの設定(後述)");
job.withApplications("EMRに入れるソフトウェア");
job.withConfigurations("EMRに入れるソフトウェアの設定");
    }

EMRに入れるソフトウェア

私の場合は、Hadoop + Spark + Gangliaを入れたかったので以下のようにしています。

 private static List<Application> buildApplications() {
     List<Application> apps = new ArrayList<>();
     apps.add(new Application().withName("Hadoop"));
     apps.add(new Application().withName("Spark"));
     apps.add(new Application().withName("Ganglia"));
     return apps;
 }

EMRに入れるソフトウェアの設定

Sparkのおまじないとして、以下を入れています。

 [{"classification":"spark","properties":{"maximizeResourceAllocation":"true"}}]

java上では以下のように記述します。

 final Configuration configuration = new Configuration();
 configuration.setClassification("spark");

 final HashMap<String, String> map = new HashMap<>();
 map.put("maximizeResourceAllocation", "true");
 configuration.setProperties(map);

 return configuration;

EMRが立ち上げるEC2インスタンスの設定、およびセキュリティ

final JobFlowInstancesConfig instanceConfig = new JobFlowInstancesConfig();
instanceConfig.withEc2KeyName("key pair名");
instanceConfig.withEc2SubnetId("EC2を立ち上げるサブネット名");
instanceConfig.withEmrManagedMasterSecurityGroup("マスターのSecurityGroup");
instanceConfig.withEmrManagedSlaveSecurityGroup("コア、タスクのSecurityGroup");
instanceConfig.withAdditionalMasterSecurityGroups("マスターへの追加のセキュリティグループ");
instanceConfig.withAdditionalSlaveSecurityGroups("コア、タスクの追加のセキュリティグループ");
instanceConfig.withKeepJobFlowAliveWhenNoSteps("Jobがない時、EMRを落とすかどうか　落とすならfalse");
instanceConfig.withInstanceGroups("インスタンス自体の設定（後述）");
instanceConfig.withTerminationProtected("インスタンスの保護");

インスタンス自体の設定に関して

final List<InstanceGroupConfig> result = new ArrayList<>();
InstanceGroupConfig masterInstance = new InstanceGroupConfig()
             .withName("マスターノード")
             .withInstanceRole(InstanceRoleType.MASTER)
             .withInstanceCount(1)
             .withInstanceType("インスタンスタイプ") //インスタンスタイプ
             .withMarket(MarketType.SPOT)
             .withBidPrice("0.14");
     result.add(masterInstance);

 InstanceGroupConfig coreInsetance = new InstanceGroupConfig()
             .withName("コアノード")
             .withInstanceRole(InstanceRoleType.CORE)
             .withInstanceCount(1)
             .withInstanceType("インスタンスタイプ")
             .withMarket(MarketType.SPOT)
             .withBidPrice("0.14");
     result.add(coreInsetance);

     InstanceGroupConfig taskInsetance = new InstanceGroupConfig()
             .withName("タスクノード")
             .withInstanceRole(InstanceRoleType.TASK)
             .withInstanceCount(4)
             .withInstanceType("インスタンスタイプ")
             .withMarket(MarketType.SPOT)
             .withBidPrice("0.14");
     result.add(taskInsetance);
     return result;

EMRで何をするかのStepの設定

私の場合は、Sparkアプリケーションで分散集計処理をするので
以下のようなコマンドをStepとして実行させます。

spark-submit --deploy-mode cluster --class my.Job s3://spark-app.jar 引数1 引数2 ...

これをJavaのコードにする場合は、以下となります。

final String[] args = {
                    "spark-submit",
                    "--deploy-mode", "cluster",
                    "--class", "my.Job",
                    "s3://spark-app.jar", 
                    "引数１", "引数２"
            };

final StepConfig stepConfig = new StepConfig()
                    .withName("step名")
                    .withActionOnFailure("終了時の振る舞い")
                    .withHadoopJarStep(new HadoopJarStepConfig()
                            .withJar("command-runner.jar")
                            .withArgs(args));

return stepConfig;

EMR起動

先ほど作ったEMRの設定をするインスタンスに,Stepを追加します

// RunJobFlowRequest job = new RunJobFlowRequest();
// ・・・

job.withSteps(steps);

以下のコードでEMRが起動します

final AmazonElasticMapReduceClient emr = new AmazonElasticMapReduceClient();
emr.setRegion(Region.getRegion(Regions.AP_NORTHEAST_1));

emr.runJobFlow(job);