Starten Sie einen Spark-Job in einem transienten EMR-Cluster mithilfe einer Lambda-Funktion

Erstellt von Dhrubajyoti Mukherjee (AWS)

Übersicht

Dieses Muster verwendet die Amazon RunJobFlow EMR-API-Aktion, um einen transienten Cluster zu starten, um einen Spark-Job von einer Lambda-Funktion aus auszuführen. Ein transienter EMR-Cluster ist so konzipiert, dass er beendet wird, sobald der Job abgeschlossen ist oder wenn ein Fehler auftritt. Ein transienter Cluster bietet Kosteneinsparungen, da er nur während der Rechenzeit ausgeführt wird, und er bietet Skalierbarkeit und Flexibilität in einer Cloud-Umgebung.

Der transiente EMR-Cluster wird mithilfe der Boto3-API und der Programmiersprache Python in einer Lambda-Funktion gestartet. Die Lambda-Funktion, die in Python geschrieben ist, bietet die zusätzliche Flexibilität, den Cluster bei Bedarf zu initiieren.

Um ein Beispiel für eine Batchberechnung und -ausgabe zu demonstrieren, startet dieses Muster einen Spark-Job in einem EMR-Cluster von einer Lambda-Funktion aus und führt eine Batch-Berechnung anhand der Beispielvertriebsdaten eines fiktiven Unternehmens durch. Die Ausgabe des Spark-Jobs wird eine Datei mit kommagetrennten Werten (CSV) in Amazon Simple Storage Service (Amazon S3) sein. Die Eingabedatendatei, die Spark-JAR-Datei, ein Codeausschnitt und eine CloudFormation AWS-Vorlage für eine Virtual Private Cloud (VPC) und AWS Identity and Access Management (IAM) -Rollen zur Ausführung der Berechnung werden als Anlage bereitgestellt.

Voraussetzungen und Einschränkungen

Voraussetzungen

Einschränkungen

Produktversionen

Architektur

Zieltechnologie-Stack

Zielarchitektur

Automatisierung und Skalierung

Um die Spark-EMR-Batchberechnung zu automatisieren, können Sie eine der folgenden Optionen verwenden.

Tools

AWS-Services

Andere Tools

Epen

Zugehörige Ressourcen

Zusätzliche Informationen

Code

"""
Copy paste the following code in your Lambda function. Make sure to change the following key parameters for the API as per your account

-Name (Name of Spark cluster)
-LogUri (S3 bucket to store EMR logs)
-Ec2SubnetId (The subnet to launch the cluster into)
-JobFlowRole (Service role for EC2)
-ServiceRole (Service role for Amazon EMR)

The following parameters are additional parameters for the Spark job itself. Change the bucket name and prefix for the Spark job (located at the bottom).

-s3://your-bucket-name/prefix/lambda-emr/SparkProfitCalc.jar (Spark jar file)
-s3://your-bucket-name/prefix/fake_sales_data.csv (Input data file in S3)
-s3://your-bucket-name/prefix/outputs/report_1/ (Output location in S3)
"""
import boto3

client = boto3.client('emr')


def lambda_handler(event, context):
    response = client.run_job_flow(
        Name='spark_job_cluster',
        LogUri='s3://your-bucket-name/prefix/logs',
        ReleaseLabel='emr-6.0.0',
        Instances={
            'MasterInstanceType': 'm5.xlarge',
            'SlaveInstanceType': 'm5.large',
            'InstanceCount': 1,
            'KeepJobFlowAliveWhenNoSteps': False,
            'TerminationProtected': False,
            'Ec2SubnetId': 'subnet-XXXXXXXXXXXXXX'
        },
        Applications=[{'Name': 'Spark'}],
        Configurations=[
            {'Classification': 'spark-hive-site',
             'Properties': {
                 'hive.metastore.client.factory.class': 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory'}
             }
        ],
        VisibleToAllUsers=True,
        JobFlowRole='EMRLambda-EMREC2InstanceProfile-XXXXXXXXX',
        ServiceRole='EMRLambda-EMRRole-XXXXXXXXX',
        Steps=[
            {
                'Name': 'flow-log-analysis',
                'ActionOnFailure': 'TERMINATE_CLUSTER',
                'HadoopJarStep': {
                    'Jar': 'command-runner.jar',
                    'Args': [
                        'spark-submit',
                        '--deploy-mode', 'cluster',
                        '--executor-memory', '6G',
                        '--num-executors', '1',
                        '--executor-cores', '2',
                        '--class', 'com.aws.emr.ProfitCalc',
                        's3://your-bucket-name/prefix/lambda-emr/SparkProfitCalc.jar',
                        's3://your-bucket-name/prefix/fake_sales_data.csv',
                        's3://your-bucket-name/prefix/outputs/report_1/'
                    ]
                }
            }
        ]
    )

IAM-Rollen und VPC-Erstellung

Um den EMR-Cluster in einer Lambda-Funktion zu starten, sind eine VPC und IAM-Rollen erforderlich. Sie können die VPC- und IAM-Rollen mithilfe der CloudFormation AWS-Vorlage im Abschnitt Anlagen dieses Musters einrichten, oder Sie können sie mithilfe der folgenden Links manuell erstellen. 

Die folgenden IAM-Rollen sind für die Ausführung von Lambda und Amazon EMR erforderlich. 

Lambda-Ausführungsrolle

Die Ausführungsrolle einer Lambda-Funktion gewährt ihr die Erlaubnis, auf AWS-Services und -Ressourcen zuzugreifen.

Servicerolle für Amazon EMR

Die Amazon EMR-Rolle definiert die zulässigen Aktionen für Amazon EMR bei der Bereitstellung von Ressourcen und der Ausführung von Service-Level-Aufgaben, die nicht im Kontext einer Amazon Elastic Compute Cloud (Amazon EC2) -Instance ausgeführt werden, die innerhalb eines Clusters ausgeführt wird. Die Servicerolle wird beispielsweise verwendet, um EC2-Instances bereitzustellen, wenn ein Cluster gestartet wird.

Servicerolle für EC2-Instances

Die Servicerolle für Cluster-EC2-Instances (auch EC2-Instance-Profil für Amazon EMR genannt) ist eine spezielle Art von Servicerolle, die jeder EC2-Instance in einem Amazon EMR-Cluster zugewiesen wird, wenn die Instance gestartet wird. Anwendungsprozesse, die auf Apache Hadoop ausgeführt werden, übernehmen diese Rolle für Berechtigungen zur Interaktion mit anderen AWS-Services.

VPC- und Subnetzerstellung

Sie können eine VPC von der VPC-Konsole aus erstellen. 

Anlagen

Um auf zusätzliche Inhalte zuzugreifen, die mit diesem Dokument verknüpft sind, entpacken Sie die folgende Datei: attachment.zip