通过使用 Amazon Graviton2 提升 EMR 的性价比

1.前言

从2020年10月开始，基于亚马逊云科技 Graviton2 的 Amazon EMR 实例逐步推出，现在已经有很多客户在不同的场景开始使用这种类型的实例。

Graviton2 处理器由亚马逊云科技使用 64 位 ARM Neoverse 内核定制，对第一代 Graviton 处理器进行了多种性能优化。这包括 7 倍的性能、4 倍的计算核心数量、每个内核 2 倍的私有内存、5 倍的内存速度和每个核心 2 倍的浮点性能。此外，Graviton2 处理器还具有全天候运行的全加密 DDR4 内存功能，且每核加密性能提高 50%。这些性能提升使得用了Graviton2 的实例成为大数据工作负载的上佳之选。

本文将向您展示在Amazon EMR大数据集群中的Graviton2 R6g实例对R5实例的性能对比测试。我们可以清楚地看到，在标准的TPC-DS基准测试环境（99个典型的SQL场景），R6g实例比之同等资源配置的R5实例性能获得提升，而且每小时单价有所下降，在这双重因素的叠加之下，在EMR大数据负载场景，选择新一代的R6g实例，具有更好的性价比。

2.环境准备

总体说明：

1.测试环境（含操作客户端，R5实例，R6g实例）：均位于新加坡区域。

2.测试EMR组件为Spark 3.1.2（对应EMR版本为6.4.0）；

3.所有EMR集群均使用Amazon Glue Data Catalog作为MDS；

4.提交测试任务方式：手工登录集群的Master节点使用Spark Shell提交；

5.测试过程中，每种场景测试多次，确保CPU（集群），内存（集群），网络（每一个Core节点），磁盘（每一个Core节点）等资源层面使用率不高于70%；

6.我们以100G的TPC-DS基准数据为测试数据，取对应查询结果做对比。

2.1环境信息

A.操作客户端（此客户端也是造数据的集群）

实例类型	vCPU	内存（GiB）	磁盘（GiB）	角色
r5.4xlarge	16	128	1500	1个Master

B.测试EMR集群：emr640-ic（基于Intel架构的R5机型）

实例类型	vCPU	内存（GiB）	磁盘（GiB）	角色
r5.4xlarge	16	128	500	1个Master
r5.8xlarge	32	256	1500	3个Core

C.测试EMR集群：emr640-ac（基于Graviton2架构的R6g机型）

实例类型	vCPU	内存（GiB）	磁盘（GiB）	角色
r6g.4xlarge	16	128	500	1个Master
r6g.8xlarge	32	256	1500	3个Core

注意：集群名称里面的ic表示intel cluster，ac表示Graviton2（arm） cluster的简称。

可能很多读者会问，为什么测试的时候只选择这一个机型？因为我们经过多轮测试以后，发现8xlarge的机型可以满足各项资源使用率不高于70%的标准。

2.2测试工具

我们使用GitHub上的TPC-DS源代码编译测试用的代码。

2.2.1 准备代码环境

登录测试客户端实例，编译如下代码（先使用sudo su – 到root用户）：

yum install gcc make flex bison byacc git jq -y 

mkdir -p /opt/tpcds && cd /opt/tpcds
git clone https://github.com/databricks/tpcds-kit.git
cd tpcds-kit/tools
make

准备造数据的对应代码：

cd /opt/tpcds

git clone https://github.com/databricks/spark-sql-perf.git
cd spark-sql-perf

wget https://github.com/sbt/sbt/releases/download/v0.13.18/sbt-0.13.18.tgz

tar zxvf sbt-0.13.18.tgz

cd build
mv sbt sbt.back
ln -s ../sbt/bin/sbt sbt
cd ..

编辑插件文件：

vi project/plugins.sbt

# 添加如下一行
# resolvers += "spark-packages" at "https://repos.spark-packages.org/"

可选操作（免得后续每一个环境都需要编译代码，我们把代码上传到S3）：

cd /opt/tpcds/spark-sql-perf/

./build/sbt +package

cd /opt
tar zcvf tpcds.tgz tpcds

aws s3 cp tpcds.tgz s3://dse-710469168206-sin/tpcds/codes/tpcds.tgz

2.2.2 准备测试数据

进入测试客户端机器

spark-shell --jars /opt/tpcds/spark-sql-perf/target/scala-2.12/spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar

用如下的方式造100g数据

import com.databricks.spark.sql.perf.tpcds.TPCDSTables

val rootDir = "s3://dse-710469168206-sin/tpcds/data100g"
val dsdgenDir = "/opt/tpcds/tpcds-kit/tools"
val scaleFactor = "100"
val format = "parquet"
val databaseName = "tpcds100g"
val sqlContext = spark.sqlContext

val tables = new TPCDSTables(sqlContext,
dsdgenDir = dsdgenDir, 
scaleFactor = scaleFactor,
useDoubleForDecimal = true, 
useStringForDate = true)

tables.genData(
location = rootDir,
format = format,
overwrite = true,
partitionTables = true, 
clusterByPartitionColumns = true, 
filterOutNullPartitionValues = false, 
tableFilter = "", 
numPartitions = 1024)

sql(s"create database $databaseName") 

tables.createExternalTables(rootDir, 
format, 
databaseName, 
overwrite = true, 
discoverPartitions = true)

tables.analyzeTables(databaseName, analyzeColumns = true)

2.3验证数据

2.3.1 控制台验证

登陆S3控制台，打开对应存储桶，查看对应数据目录内容（我们造了多个数据库，有10g的，有1000g的，但是此处我们只测试100g的场景）。

登录Glue控制台，找到对应数据库，查看对应数据库和表。

2.3.2 代码验证

使用如下代码执行查询并输出结果

spark-shell --jars /opt/tpcds/spark-sql-perf/target/scala-2.12/spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar


import com.databricks.spark.sql.perf.tpcds.TPCDS

val sqlContext = spark.sqlContext
val tpcds = new TPCDS(sqlContext)

// name of database with TPCDS data.
val databaseName = "tpcds10g" 
sql(s"use $databaseName")

// place to write results
val resultLocation = "s3://dse-710469168206-sin/result/tpcds100g/ic"

// how many iterations of queries to run.
val iterations = 1 

// queries to run.
val queries = tpcds.tpcds2_4Queries

// timeout per query, in seconds.
val timeout = 7200
 
//run experiment
val experiment = tpcds.runExperiment(queries, iterations = iterations,resultLocation = resultLocation,forkThread = true)

// wait for finish
experiment.waitForFinish(timeout)

验证运行99个查询后的结果输出（记得在这个master上执行aws configure配置ak/sk并设置profile为default）

pyspark

# 然后执行
from pyspark.sql import Row
resultLocation = "s3a://dse-710469168206-sin/result/tpcds100g/ic"

result = spark.read.format("json").load(resultLocation + "/timestamp=1631844109600").select("results.name","results.executionTime")

def getRowList(row):
            lst = []
            for i in range(0, len(row.name)):
                        lst.append(Row(name=row.name[i],executionTime=row.executionTime[i]))
            return lst

resultDF = result.rdd.flatMap(lambda row: getRowList(row)).toDF().show(104)

3.对比测试过程

我们在部署Intel和Graviton2集群时均做了如下所示的Spark优化配置（在部署EMR集群时作为参数带入）：

[
    {
        "Classification": "spark-defaults",
        "Properties": {
            "spark.executor.cores": "5",
            "spark.driver.cores": "16",
            "spark.executor.memory": "32g",
            "spark.driver.memory": "32g",
            "spark.executor.instances": "16",
            "spark.default.parallelism": "144",
            "spark.sql.shuffle.partitions": "1024",
            "spark.network.timeout": "900s",
            "spark.driver.maxResultSize": "0",
            "spark.executor.heartbeatInterval": "60s",
            "spark.dynamicAllocation.enabled": "true",
            "spark.yarn.scheduler.reporterThread.maxFailures": "2000",
            "spark.sql.debug.maxToStringFields": "2000",
            "spark.sql.autoBroadcastJoinThreshold": "-1"
        }
    },
    {
        "Classification": "yarn-site",
        "Properties": {
            "yarn.nodemanager.vmem-check-enabled": "false",
            "yarn.nodemanager.pmem-check-enabled": "false"
        }
    }
]

并使用之前单节点测试集群的Glue Data Catalog作为MDS。

3.1Intel集群测试

我们测试的过程中发现，不管测试多少轮，其实结果都类似，所以我们取3次正常的数据作为结果。

3.1.1准备测试代码

启动集群后，登录Master和Core节点复制代码(sudo su -)

cd /opt

yum install gcc make flex bison byacc git jq -y

aws s3 cp s3://dse-710469168206-sin/tpcds/codes/tpcds.tgz tpcds.tgz

tar zxvf tpcds.tgz

3.1.2执行查询测试

登录集群Master，通过如下方式提交查询测试任务（此为100g数据源的代码，其他的请酌情修改）

spark-shell --jars /opt/tpcds/spark-sql-perf/target/scala-2.12/spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar

import com.databricks.spark.sql.perf.tpcds.TPCDS

val sqlContext = spark.sqlContext
val tpcds = new TPCDS(sqlContext)

val databaseName = "tpcds100g" 
sql(s"use $databaseName")

// place to write results
val resultLocation = "s3://dse-710469168206-sin/result/tpcds100g/ic"

// how many iterations of queries to run.
val iterations = 1

// queries to run.
val queries = tpcds.tpcds2_4Queries

// timeout per query, in seconds.
val timeout = 7200

//run experiment
val experiment = tpcds.runExperiment(queries, iterations = iterations,resultLocation = resultLocation,forkThread = true)

// wait for finish
experiment.waitForFinish(timeout)

注意查询结束后输出的类似如下的结果地址

s3://dse-710469168206-sin/result/tpcds100g/ic/timestamp=1632029414066

多执行几次，并打开Ganglia查看对应监控信息，以99个SQL无任何错误，使用率不超过70%为判断依据。

3.1.3监控情况

从监控数据显示，系统CPU和内存使用几乎接近如下所示（截图仅供参考），我们尽可能保证在没有任何异常的情况下，CPU和内存使用率均保持在50%左右，避免因某些因素导致测试结果失去公平性。

CPU使用情况截图：

内存使用情况截图：

网络使用情况截图：

十五分钟集群负载：

十五分钟单机负载：

只要碰到监控异常或者99个SQL执行结果异常的数据就剔除不要。

3.2Graviton2集群测试

3.2.1准备测试代码

启动集群后，登录Master和Core节点复制代码(sudo su -)

cd /opt

yum install gcc make flex bison byacc git jq -y

aws s3 cp s3://dse-710469168206-sin/tpcds/codes/tpcds.tgz tpcds.tgz

tar zxvf tpcds.tgz

3.2.2执行查询测试

登录集群Master，通过如下方式提交查询测试任务（此为100g数据源的代码，其他的请酌情修改）

spark-shell --jars /opt/tpcds/spark-sql-perf/target/scala-2.12/spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar

import com.databricks.spark.sql.perf.tpcds.TPCDS

val sqlContext = spark.sqlContext
val tpcds = new TPCDS(sqlContext)

val databaseName = "tpcds100g" 
sql(s"use $databaseName")

// place to write results
val resultLocation = "s3://dse-710469168206-sin/result/tpcds100g/ac"

// how many iterations of queries to run.
val iterations = 1

// queries to run.
val queries = tpcds.tpcds2_4Queries

// timeout per query, in seconds.
val timeout = 7200

//run experiment
val experiment = tpcds.runExperiment(queries, iterations = iterations,resultLocation = resultLocation,forkThread = true)

// wait for finish
experiment.waitForFinish(timeout)

注意查询结束后输出的类似如下的结果地址

s3://dse-710469168206-sin/result/tpcds100g/ac/timestamp=1632029414066

多执行几次，并打开Ganglia查看对应监控信息，以99个SQL无任何错误，使用率不超过70%为判断依据。

3.2.3监控情况

CPU使用情况截图：

内存使用情况截图：

网络使用情况截图：

十五分钟集群负载：

十五分钟单机负载：

只要碰到监控异常或者99个SQL执行结果异常的数据就剔除不要。

3.3测试结果

3.3.1取结果的方式

登录单Master的集群的Master节点，查询对应结果

pyspark

# 然后执行
from pyspark.sql import Row
resultLocation = "s3a://dse-710469168206-sin/result/tpcds100g/ac"

result = spark.read.format("json").load(resultLocation + "/timestamp=1632029414066").select("results.name","results.executionTime")

def getRowList(row):
            lst = []
            for i in range(0, len(row.name)):
                        lst.append(Row(name=row.name[i],executionTime=row.executionTime[i]))
            return lst


resultDF = result.rdd.flatMap(lambda row: getRowList(row)).toDF().show(104)

3.3.2结果汇总

Intel集群（emr640-ic）测试数据：

Graviton2集群（emr640-ac）测试数据

4.对比分析

因为一共有99个SQL，所以没办法简单的通过单个SQL（不管是执行时间最长的还是最短的或者中位线）的执行时间来评判性能，所以我们先把多次（此处为3次）的执行结果取平均值，然后根据99个SQL的执行结果的评价值算出来百分比，根据平均延时（秒）和百分比的乘积算出总数，并各自基于总数折算单成本（此处的成本不是指钱，是指计算成本），最后加和得出一个总值作为测试对比结果。

4.1性能对比

按照上述计算方式我们算出的性能成本如下：

平台	总时间成本	单位成本	比例
Intel	1102.26	30.5217
Graviton2	1067.36	26.2709	16.18%

从结果看，基于TPC-DS的100G数据和99个SQL查询，Graviton2（ARM）架构的EMR集群性能同比约优胜16%。

4.2性价比对比

根据官网的报价（https://thinkwithwp.com/cn/emr/pricing/），我们选择新加坡区域，列表价如下：

类型	EC2列表价($)	EMR列表价($)	合计	比例
r5.4xlarge	1.2160	0.2520	1.4680
r6g.4xlarge	0.9728	0.2016	1.1744	25.00%

r5.8xlarge	2.4320	0.2700	2.7020
r6g.8xlarge	1.9456	0.4032	2.3488	15.03%

此处我们不考虑预留实例，企业折扣等优惠方式，纯从列表价看，合并后（包括EC2实例和EMR）的总单价，Graviton2架构明显优于Intel架构。

4.3结论

细心的读者可能会留意到，我们并没有额外再针对Graviton2（ARM）架构重新编译代码，同样的代码在两套架构上完全通用。当然，因为我们全部用的是Spark SQL方式所以架构差异较少，如果是客户的实际业务环境还是可能会有差异的，建议读者朋友们以测试结果为准。

从我们的测试结果看，从Intel架构迁移到Graviton2（ARM）架构，不仅可以获得性能的提升，还可以获得更好的性价比。

亚马逊AWS官方博客