added support to run in AWS EMR

Author: michaelkamprath

aws-emr/build-test-data-emr.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -e
+
+aws emr create-cluster --name "Pyspark Benchmark - Generate Data" \
+--release-label emr-5.29.0 \
+--applications Name=Spark \
+--log-uri s3://your-s3-bucket/logs/ \
+--ec2-attributes KeyName=your-key-pair \
+--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=r5d.xlarge,BidPrice=OnDemandPrice InstanceGroupType=CORE,InstanceCount=6,InstanceType=r5d.2xlarge,BidPrice=OnDemandPrice \
+--bootstrap-actions Path=s3://your-s3-bucket/emr_bootstrap.sh \
+--steps Type=Spark,Name="Pyspark Benchmark - Generate Data",\
+ActionOnFailure=CONTINUE,\
+Args=[--deploy-mode,cluster,--master,yarn,s3://your-s3-bucket/jobs/generate-data.py,s3://your-s3-bucket/data/,-r,2000000000,-p,1000] \
+--use-default-roles \
+--auto-terminate
+

aws-emr/emr_bootstrap.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+sudo pip install -U \
+    matplotlib \
+    pandas 

aws-emr/run-benchmarks-emr.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+
+set -e
+
+aws emr create-cluster --name "Pyspark Benchmark - Shuffle" \
+--release-label emr-5.29.0 \
+--applications Name=Spark \
+--log-uri s3://your-s3-bucket/logs/ \
+--ec2-attributes KeyName=your-key-pair \
+--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=r5d.xlarge,BidPrice=OnDemandPrice InstanceGroupType=CORE,InstanceCount=6,InstanceType=r5d.2xlarge,BidPrice=OnDemandPrice \
+--bootstrap-actions Path=s3://your-s3-bucket/emr_bootstrap.sh \
+--steps Type=Spark,Name="Pyspark Benchmark - Shuffle",\
+ActionOnFailure=CONTINUE,\
+Args=[--deploy-mode,cluster,--master,yarn,s3://your-s3-bucket/jobs/benchmark-shuffle.py,s3://your-s3-bucket/data/,-r,250,-n,'pyspark-benchmark-shuffle',-o,s3://your-s3-bucket/results/pyspark-shuffle] \
+--use-default-roles \
+--auto-terminate
+
+
+aws emr create-cluster --name "Pyspark Benchmark - CPU" \
+--release-label emr-5.29.0 \
+--applications Name=Spark \
+--log-uri s3://your-s3-bucket/logs/ \
+--ec2-attributes KeyName=your-key-pair \
+--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=r5d.xlarge,BidPrice=OnDemandPrice InstanceGroupType=CORE,InstanceCount=6,InstanceType=r5d.2xlarge,BidPrice=OnDemandPrice \
+--bootstrap-actions Path=s3://your-s3-bucket/emr_bootstrap.sh \
+--steps Type=Spark,Name="Pyspark Benchmark - CPU",\
+ActionOnFailure=CONTINUE,\
+Args=[--deploy-mode,cluster,--master,yarn,s3://your-s3-bucket/jobs/benchmark-cpu.py,s3://your-s3-bucket/data/,-s,25000000000,-p,1000,-n,'pyspark-benchmark-cpu',-o,s3://your-s3-bucket/results/pyspark-cpu] \
+--use-default-roles \
+--auto-terminate

benchmark-cpu.py

@@ -63,6 +63,14 @@ def parseArguments():
             dest='appName',
             help='The name given this PySpark job'
         )
+    arguments.add_argument(
+            '-o', '--results-output',
+            metavar='results-file-path',
+            type=str,
+            default=None,
+            dest='results_output_file',
+            help='The file path to place the results output'
+        )
     return arguments.parse_args()
 
 def benchmarkSHA256(df, jobLogger):
@@ -187,6 +195,27 @@ def main():
     joblogger.info('')
     joblogger.info('****************************************************************************')
 
+    if args.results_output_file is not None:
+        joblogger.info('')
+        joblogger.info('Writing results to {0}'.format(args.results_output_file))
+    
+        results_list = [
+            ('sha-512',sha256_time),
+            ('calc-pi-python-udf',calcPi_time),
+            ('calc-pi-dataframe',calcPi_DF_time),
+        ]
+    
+        results_schema = T.StructType([
+            T.StructField("test", T.StringType()),
+            T.StructField("seconds", T.DoubleType())
+        ])
+        results_df = spark.createDataFrame(results_list, schema=results_schema).coalesce(1)
+        results_df.write.csv(
+            args.results_output_file,
+            header=True,
+            mode='overwrite'
+        )
+
 
 if __name__ == '__main__':
     main()

benchmark-shuffle.py

@@ -54,6 +54,14 @@ def parseArguments():
             dest='repartitions',
             help='The number of partitions to use in repartition benchmark'
         )
+    arguments.add_argument(
+            '-o', '--results-output',
+            metavar='results-file-path',
+            type=str,
+            default=None,
+            dest='results_output_file',
+            help='The file path to place the results output'
+        )
     return arguments.parse_args()
 
 def benchmarkGroupBy(df, jobLogger):
@@ -221,5 +229,27 @@ def main():
     joblogger.info('')
     joblogger.info('**********************************************************************')
 
+    if args.results_output_file is not None:
+        joblogger.info('')
+        joblogger.info('Writing results to {0}'.format(args.results_output_file))
+
+        results_list = [
+            ('group-by',groupBy_time),
+            ('repartition',repartition_time),
+            ('inner-join',innerJoin_time),
+            ('broadcast-inner-join',broadcastInnerJoin_time),
+        ]
+    
+        results_schema = T.StructType([
+            T.StructField("test", T.StringType()),
+            T.StructField("seconds", T.DoubleType())
+        ])
+        results_df = spark.createDataFrame(results_list, schema=results_schema).coalesce(1)
+        results_df.write.csv(
+            args.results_output_file,
+            header=True,
+            mode='overwrite'
+        )
+
 if __name__ == '__main__':
     main()