Luncurkan pekerjaan Spark di cluster EMR sementara menggunakan fungsi Lambda

Dhrubajyoti Mukherjeee, Amazon Web Services

Ringkasan

Pola ini menggunakan aksi Amazon EMR RunJobFlow API untuk meluncurkan klaster sementara untuk menjalankan pekerjaan Spark dari fungsi Lambda. Cluster EMR sementara dirancang untuk dihentikan segera setelah pekerjaan selesai atau jika terjadi kesalahan. Cluster transien memberikan penghematan biaya karena hanya berjalan selama waktu komputasi, dan memberikan skalabilitas dan fleksibilitas dalam lingkungan cloud.

Cluster EMR transien diluncurkan menggunakan API Boto3 dan bahasa pemrograman Python dalam fungsi Lambda. Fungsi Lambda, yang ditulis dengan Python, memberikan fleksibilitas tambahan untuk memulai cluster saat dibutuhkan.

Untuk mendemonstrasikan komputasi dan output batch sampel, pola ini akan meluncurkan pekerjaan Spark di cluster EMR dari fungsi Lambda dan menjalankan komputasi batch terhadap contoh data penjualan perusahaan fiksi. Output dari pekerjaan Spark akan menjadi file nilai yang dipisahkan koma (CSV) di Amazon Simple Storage Service (Amazon S3). File data input, file Spark .jar, cuplikan kode, dan CloudFormation template AWS untuk peran virtual private cloud (VPC) dan AWS Identity and Access Management (IAM) untuk menjalankan komputasi disediakan sebagai lampiran.

Prasyarat dan batasan

Prasyarat

Batasan

Versi produk

Arsitektur

Tumpukan teknologi target

Arsitektur target

Otomatisasi dan skala

Untuk mengotomatiskan komputasi batch Spark-EMR, Anda dapat menggunakan salah satu dari opsi berikut.

Alat

Layanan AWS

Alat lainnya

Epik

Sumber daya terkait

Informasi tambahan

Kode

"""
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/'
                    ]
                }
            }
        ]
    )

Peran IAM dan pembuatan VPC

Untuk meluncurkan cluster EMR dalam fungsi Lambda, diperlukan peran VPC dan IAM. Anda dapat mengatur peran VPC dan IAM dengan menggunakan CloudFormation templat AWS di bagian Lampiran pola ini, atau Anda dapat membuatnya secara manual menggunakan tautan berikut. 

Peran IAM berikut diperlukan untuk menjalankan Lambda dan Amazon EMR. 

Peran eksekusi Lambda

Peran eksekusi fungsi Lambda memberinya izin untuk mengakses layanan dan sumber daya AWS.

Peran layanan untuk Amazon EMR

Peran Amazon EMR mendefinisikan tindakan yang diizinkan untuk Amazon EMR saat menyediakan sumber daya dan melakukan tugas tingkat layanan yang tidak dilakukan dalam konteks instans Amazon Elastic Compute Cloud (Amazon) yang berjalan dalam klaster. EC2 Misalnya, peran layanan digunakan untuk menyediakan EC2 instance saat klaster diluncurkan.

Peran layanan untuk EC2 instance

Peran layanan untuk EC2 instance klaster (juga disebut profil EC2 instans untuk Amazon EMR) adalah jenis peran layanan khusus yang ditetapkan ke EC2 setiap instance dalam klaster EMR Amazon saat instance diluncurkan. Proses aplikasi yang berjalan di atas Apache Hadoop mengambil peran ini sebagai izin untuk berinteraksi dengan layanan AWS lainnya.

VPC dan pembuatan subnet

Anda dapat membuat VPC dari konsol VPC. 

Lampiran

Untuk mengakses konten tambahan yang terkait dengan dokumen ini, unzip file berikut: attachment.zip