This is a guest blog by Chengzhi Zhao with an original blog source.

Traditionally, if you want to run a single Spark job on EMR, you might follow the steps: launching a cluster, running the job which reads data from storage layer like S3, performing transformations within RDD/Dataframe/Dataset, finally, sending the result back to S3. You end up having something like this:

If we add more Spark jobs across multiple clusters, you could have something like this.

There can be more use cases, for example, sometimes we need to store the intermediate result of Spark job A, and used the intermediate result as input for Spark job B; sometimes you would have multiple Spark jobs read data from the same dataset multiple times. As for now, each of the Spark jobs have to read input data from disk then process it. What if we can read input from memory? Alluxio is a solution for it.

Alluxio is the storage underneath that usually collocates with the computation frameworks, so that Alluxio can provide fast storage, facilitating data sharing and locality between jobs, regardless of whether they are running on the same computation engine.

I will show you how to set up Alluxio 1.8.1 on EMR 5.17 with a bootstrap script and compare the data loading performance.

Some more detail is described in the following repository: https://github.com/ChengzhiZhao/Alluxio-EMR-bootstrap

STEP 1: set up a bootstrap script

Use bootstrap.sh script includes download Alluxio 1.8.1 and setup required permissions on EMR

STEP 2: add the required configuration.

[ { "Classification": "core-site", "Properties": { "fs.alluxio.impl": "alluxio.hadoop.FileSystem", "fs.AbstractFileSystem.alluxio.impl": "alluxio.hadoop.AlluxioFileSystem" } }, { "Classification": "spark-defaults", "Properties": { "spark.driver.extraClassPath": ":/usr/lib/hadoop-lzo/lib/*:/usr/lib/hadoop/hadoop-aws.jar:/usr/share/aws/aws-java-sdk/*:/usr/share/aws/emr/emrfs/conf:/usr/share/aws/emr/emrfs/lib/*:/usr/share/aws/emr/emrfs/auxlib/*:/usr/share/aws/emr/security/conf:/usr/share/aws/emr/security/lib/*:/opt/alluxio-1.8.1-hadoop-2.8/client/alluxio-1.8.1-client.jar", "spark.executor.extraClassPath": ":/opt/alluxio-1.8.1-hadoop-2.8/client/alluxio-1.8.1-client.jar" } }]

STEP 3: launch the cluster, take a break and relax ☕️

STEP 4: ssh to EMR cluster and perform copy test data to make sure Alluxio is running

cd /opt/alluxio-1.8.1-hadoop-2.8 # copy files to alluxio bin/alluxio fs copyFromLocal LICENSE /Input # Copied file:///opt/alluxio-1.8.1-hadoop-2.8/LICENSE to /Input # verify files by listing alluxio bin/alluxio fs ls / # 26847 NOT_PERSISTED 02-18-2019 19:22:28:025 100% /Input

You can also check Alluxio UI by going to {Master public DNS}:19999

Click “Browse” and you should see Input file there with default 128 MB as block size.

STEP 5: Use Spark shell to read the “Input” file from Alluxio

#launch spark-shell spark-shell #read data from alluxio and use the DNS -- in this case, ip-10-192-4-226.ec2.internal val s = sc.textFile("alluxio://ip-10-192-4-226.ec2.internal:19998/Input") #s: org.apache.spark.rdd.RDD[String] = alluxio://ip-10-192-4-226.ec2.internal:19998/Input MapPartitionsRDD[1] at textFile at <console>:24 s.count res0: Long = 476

STEP 6: Read and compare as a real example

Note: You can use any test_data_set with Spark, I just pick a test Avro file online.

# read test_data_set val df=spark.read.format("com.databricks.spark.avro").load("s3://test_bucket/test_data_set/date=2019-04-18") spark.time(df.count) #Time taken: 29609 ms #res0: Long = 86731200 # write df to alluxio spark.time(df.write.parquet("alluxio://ip-10-192-4-226.ec2.internal:19998/data.test_data_set")) #Time taken: 45775 ms # first execution spark.time(spark.read.parquet("alluxio://ip-10-192-4-226.ec2.internal:19998/data.test_data_set").count) #Time taken: 13477 ms #res3: Long = 86731200 # second execution spark.time(spark.read.parquet("alluxio://ip-10-192-4-226.ec2.internal:19998/data.test_data_set").count) #Time taken: 372 ms #res4: Long = 86731200 # restart with another spark-shell spark.time(spark.read.parquet("alluxio://ip-10-192-4-226.ec2.internal:19998/data.test_data_set").count) #Time taken: 9606 ms #res0: Long = 86731200

As you can see, the second execution is faster, it is similar as we perform cache. But if we close and open another spark-shell, since we read data from Alluxio and data is kept in memory, it would be faster. If you are interested, please refer to this great talk — Best Practices for Using Alluxio with Apache Spark which talks benchmark for Alluxio compared with Spark.

Performance Comparison (Count job with reading from Disk vs. Alluxio)

pros: 1) faster speed to read data as df than from S3
cons: 1) Alluxio persist data first in memory to achieve speed, so spark job could have less memory to run jobs. 2) there are overhead time to write df to Allluxio first

What’s not been covered is Alluxio is also a great storage sharing layer for multiple jobs read and write data. You could have a Flink job that writes to Alluxio and later used by Spark, there are more interesting topics on it. This post focus on how to setup Alluxio 1.8.1 on EMR and run simple test data on it.