解决方案：如何在 Amazon EMR Serverless 上执行纯 SQL 文件？

博主历时三年精心创作的《大数据平台架构与原型实现：数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行，点击《重磅推荐：建大数据平台太难了！给我发个工程原型吧！》了解图书详情，京东购书链接：https://item.jd.com/12677623.html，扫描左侧二维码进入京东手机购书页面。

长久已来，SQL以其简单易用、开发效率高等优势一直是ETL的首选编程语言，在构建数据仓库和数据湖的过程中发挥着不可替代的作用。Hive和Spark SQL也正是立足于这一点，才在今天的大数据生态中牢牢占据着主力位置。在常规的Spark环境中，开发者可以使用spark-sql命令直接执行SQL文件，这是一项看似平平无奇实则非常重要的功能：一方面，这一方式极大地降低了Spark的使用门槛，用户只要会写SQL就可以使用Spark；另一方面，通过命令行驱动SQL文件的执行可以极大简化SQL作业的提交工作，使得作业提交本身被“代码化”，为大规模工程开发和自动化部署提供了便利。

但遗憾的是，Amazon EMR Serverless 未能针对执行SQL文件提供原生支持，用户只能在Scala/Python代码中嵌入SQL语句，这对于倚重纯SQL开发数仓或数据湖的用户来说并不友好。为此，我们专门开发了一组用于读取、解析和执行SQL文件的工具类，借助这组工具类，用户可以在 Amazon EMR Serverless 上直接执行SQL文件，本文将详细介绍一下这一方案。

1. 方案设计

项目中的com.github.emr.serverless.SparkSqlJob类即为通用的SQL作业类，该类接受两个可选参数，分别是：

该方案具备如下特性：

① 允许单一SQL文件包含多条SQL语句
② 允许在SQL文件中使用${K1},${K2},…的形式定义变量，并在执行作业时使用K1=V1,K2=V2,...形式的参数进行变量赋值
③ 支持Java文件系统通配符，可一次执行多个SQL文件

下面，我们将分别在AWS控制台和命令行两种环境下介绍并演示如何使用该项目的工具类提交纯SQL作业。

2. 实操演示

2.1. 环境准备

接下来要准备提交作业所需的Jar包和SQL文件。首先在S3上创建一个存储桶，本文使用的桶取名：my-spark-sql-job（当您在自己的环境中操作时请注意替换桶名），然后从 [ 此处 ] 下载编译好的 emr-serverless-utils.jar包并上传至s3://my-spark-sql-job/jars/目录下：

在演示过程中还将使用到5个SQL示例文件，从 [ 此处 ] 下载解压后上传至s3://my-spark-sql-job/sqls/目录下：

2.2. 在控制台上提交纯SQL文件作业

2.2.1. 执行单一SQL文件

① Script location：设定为此前上传的Jar包路径 s3://my-spark-sql-job/jars/emr-serverless-utils-1.0.jar
② Main class：设定为 com.github.emr.serverless.SparkSqlJob
③ Script arguments：设定为 ["--sql-files","s3://my-spark-sql-job/sqls/drop-tables.sql"]

至于其他选项，无需特别设定，保持默认配置即可，对于在生产环境中部署的作业，您可以结合自身作业的需要灵活配置，例如Spark Driver/Executor的资源分配等。需要提醒的是：通过控制台创建的作业默认会启用Glue Data Catalog（即：Additional settings -> Metastore configuration -> Use AWS Glue Data Catalog 默认是勾选的），为了方便在Glue和Athena中检查SQL脚本的执行结果，建议您不要修改此项默认配置。

上述配置描述了这样一项工作：以s3://my-spark-sql-job/jars/emr-serverless-utils-1.0.jar中的com.github.emr.serverless.SparkSqlJob作为主类，提起一个Spark作业。其中["--sql-files","s3://my-spark-sql-job/sqls/drop-tables.sql"]是传递给SparkSqlJob的参数，用于告知作业所要执行的SQL文件位置。本次作业执行的SQL文件只有三条简单的DROP TABLE语句，是一个基础示例，用以展示工具类执行单一文件内多条SQL语句的能力。

2.2.2. 执行带自定义参数的SQL文件

["--sql-files","s3://my-spark-sql-job/sqls/create-tables.sql","--sql-params","APP_S3_HOME=s3://my-spark-sql-job"]

如下图所示：

这次的作业设定除了使用--sql-files参数指定了SQL文件外，还通过--sql-params参数为SQL中出现的用户自定义变量进行了赋值。根据此前的介绍，APP_S3_HOME=s3://my-spark-sql-job是一个“Key=Value”字符串，其含义是将值s3://my-spark-sql-job赋予了变量APP_S3_HOME，SQL中所有出现${APP_S3_HOME}的地方都将被s3://my-spark-sql-job所替代。查看create-tables.sql文件，在建表语句的LOCATION部分可以发现自定义变量${APP_S3_HOME}：

CREATE EXTERNAL TABLE IF NOT EXISTS ORDERS (
    ... ...
)
... ...
LOCATION '${APP_S3_HOME}/data/orders/';

当SparkSqlJob读取该SQL文件时，会根据键值对字符串APP_S3_HOME=s3://my-spark-sql-job将SQL文件中所有的${APP_S3_HOME}替换为s3://my-spark-sql-job，实际执行的SQL将变为：

CREATE EXTERNAL TABLE IF NOT EXISTS ORDERS (
    ... ...
)
... ...
LOCATION 's3://my-spark-sql-job/data/orders/';

提交作业并执行完毕后，可登录Athena控制台，查看数据表是否创建成功。

2.2.3. 使用通配符执行多个文件

["--sql-files","s3://my-spark-sql-job/sqls/insert-into-*.sql"]

如下图所示：

这次作业的--sql-files参数使用了路径通配符，insert-into-*.sql将同时匹配insert-into-orders.sql和insert-into-customers.sql两个SQL文件，它们将分别向ORDERS和CUSTOMERS两张表插入多条记录。执行完毕后，可以可登录Athena控制台，查看数据表中是否有数据产生。

2.2.4. 一个复合示例

["--sql-files","s3://my-spark-sql-job/sqls/select-*.sql","--sql-params","APP_S3_HOME=s3://my-spark-sql-job,CUST_CITY=NEW YORK,ORD_DATE=2008-07-15"]

如下图所示：

![emr-serverless-snapshot-4.jpg-150.8kB][6]

本次作业的--sql-files参数使用路径通配符select-*.sql匹配select-tables.sql文件，该文件中存在三个用户自定义变量，分别是${APP_S3_HOME}、${CUST_CITY}、${ORD_DATE}：

CREATE EXTERNAL TABLE ORDERS_CUSTOMERS
    ... ...
    LOCATION '${APP_S3_HOME}/data/orders_customers/'
AS SELECT
    ... ...
WHERE
    C.CUST_CITY = '${CUST_CITY}' AND
    O.ORD_DATE = CAST('${ORD_DATE}' AS DATE);

--sql-params参数为这三个自定义变量设置了取值，分别是：APP_S3_HOME=s3://my-spark-sql-job，CUST_CITY=NEW YORK，ORD_DATE=2008-07-15，于是上述SQL将被转化为如下内容去执行：

CREATE EXTERNAL TABLE ORDERS_CUSTOMERS
    ... ...
    LOCATION 's3://my-spark-sql-job/data/orders_customers/'
AS SELECT
    ... ...
WHERE
    C.CUST_CITY = 'NEW YORK' AND
    O.ORD_DATE = CAST('2008-07-15' AS DATE);

至此，通过控制台提交纯SQL文件作业的所有功能演示完毕。

2.3. 通过命令行提交纯SQL文件作业

本文使用命令行提交EMR Serverless作业的方式遵循了《最佳实践：如何优雅地提交一个 Amazon EMR Serverless 作业？》一文给出的最佳实践。首先，登录一个安装了AWS CLI并配置有用户凭证的Linux环境（建议使用Amazon Linux2），先使用命令sudo yum -y install jq安装操作json文件的命令行工具：jq（后续脚本会使用到它），然后完成如下前期准备工作：

① 创建或选择一个作业专属工作目录和S3存储桶
② 创建或选择一个EMR Serverless Execution Role
③ 创建或选择一个EMR Serverless Application

接下来将所有环境相关变量悉数导出（请根据您的AWS账号和本地环境替换命令行中的相应值）：

export APP_NAME='change-to-your-app-name'
export APP_S3_HOME='change-to-your-app-s3-home'
export APP_LOCAL_HOME='change-to-your-app-local-home'
export EMR_SERVERLESS_APP_ID='change-to-your-application-id'
export EMR_SERVERLESS_EXECUTION_ROLE_ARN='change-to-your-execution-role-arn'

以下是一份示例：

export APP_NAME='my-spark-sql-job'
export APP_S3_HOME='s3://my-spark-sql-job'
export APP_LOCAL_HOME='/home/ec2-user/my-spark-sql-job'
export EMR_SERVERLESS_APP_ID='00fbfel40ee59k09'
export EMR_SERVERLESS_EXECUTION_ROLE_ARN='arn:aws:iam::123456789000:role/EMR_SERVERLESS_ADMIN'

《最佳实践：如何优雅地提交一个 Amazon EMR Serverless 作业？》一文提供了多个操作Job的通用脚本，都非常实用，本文也会直接复用这些脚本，但是由于我们需要多次提交且每次的参数又有所不同，为了便于使用和简化行文，我们将原文中的部分脚本封装为一个Shell函数，取名为submit-spark-sql-job：

submit-spark-sql-job() {
    sqlFiles="$1"
    sqlParams="$2"
    cat << EOF > $APP_LOCAL_HOME/start-job-run.json
{
    "name":"my-spark-sql-job",
    "applicationId":"$EMR_SERVERLESS_APP_ID",
    "executionRoleArn":"$EMR_SERVERLESS_EXECUTION_ROLE_ARN",
    "jobDriver":{
        "sparkSubmit":{
        "entryPoint":"$APP_S3_HOME/jars/emr-serverless-utils-1.0.jar",
        "entryPointArguments":[
            $([[ -n "$sqlFiles" ]] && echo "\"--sql-files\", \"$sqlFiles\"")
            $([[ -n "$sqlParams" ]] && echo ",\"--sql-params\", \"$sqlParams\"")
        ],
         "sparkSubmitParameters":"--class com.github.emr.serverless.SparkSqlJob --conf spark.hadoop.hive.metastore.client.factory.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
        }
   },
   "configurationOverrides":{
        "monitoringConfiguration":{
            "s3MonitoringConfiguration":{
                "logUri":"$APP_S3_HOME/logs"
            }
        }
   }
}
EOF
    jq . $APP_LOCAL_HOME/start-job-run.json
    export EMR_SERVERLESS_JOB_RUN_ID=$(aws emr-serverless start-job-run \
        --no-paginate --no-cli-pager --output text \
        --name my-spark-sql-job \
        --application-id $EMR_SERVERLESS_APP_ID \
        --execution-role-arn $EMR_SERVERLESS_EXECUTION_ROLE_ARN \
        --execution-timeout-minutes 0 \
        --cli-input-json file://$APP_LOCAL_HOME/start-job-run.json \
        --query jobRunId)
    now=$(date +%s)sec
    while true; do
        jobStatus=$(aws emr-serverless get-job-run \
                        --no-paginate --no-cli-pager --output text \
                        --application-id $EMR_SERVERLESS_APP_ID \
                        --job-run-id $EMR_SERVERLESS_JOB_RUN_ID \
                        --query jobRun.state)
        if [ "$jobStatus" = "PENDING" ] || [ "$jobStatus" = "SCHEDULED" ] || [ "$jobStatus" = "RUNNING" ]; then
            for i in {0..5}; do
                echo -ne "\E[33;5m>>> The job [ $EMR_SERVERLESS_JOB_RUN_ID ] state is [ $jobStatus ], duration [ $(date -u --date now-$now +%H:%M:%S) ] ....\r\E[0m"
                sleep 1
            done
        else
            printf "The job [ $EMR_SERVERLESS_JOB_RUN_ID ] is [ $jobStatus ]%50s\n\n"
            break
        fi
    done
}

该函数接受两个位置参数：

① 第一位置上的参数用于指定SQL文件路径，其值会传递给SparkSqlJob的--sql-files
② 第二位置上的参数用于指定SQL文件中的用户自定义变量，其值会传递给SparkSqlJob的--sql-params

函数中使用的Jar包和SQL文件与《2.1. 环境准备》一节准备的Jar包和SQL文件一致，所以使用脚本提交作业前同样需要完成2.1节的环境准备工作。接下来，我们就使用该函数完成与2.2节一样的操作。

2.3.1. 执行单一SQL文件

submit-spark-sql-job "$APP_S3_HOME/sqls/drop-tables.sql"

2.3.2. 执行带自定义参数的SQL文件

submit-spark-sql-job "$APP_S3_HOME/sqls/create-tables.sql" "APP_S3_HOME=$APP_S3_HOME"

2.3.3. 使用通配符执行多个文件

submit-spark-sql-job "$APP_S3_HOME/sqls/insert-into-*.sql"

2.3.4. 一个复合示例

submit-spark-sql-job "$APP_S3_HOME/sqls/select-tables.sql" "APP_S3_HOME=$APP_S3_HOME,CUST_CITY=NEW YORK,ORD_DATE=2008-07-15"

3. 在源代码中调用工具类

① 将emr-serverless-utils-1.0.jar加载到你的类路径中
② 声明隐式类型转换
③ 在spark上直接调用execSqlFile()文章来源地址https://www.toymoban.com/news/detail-653488.html

# 初始化SparkSession及其他操作
...

# 声明隐式类型转换
import com.github.emr.serverless.SparkSqlSupport._

# 在spark上直接调用execSqlFile()
spark.execSqlFile("s3://YOUR/XXX.sql")

# 在spark上直接调用execSqlFile()
spark.execSqlFile("s3://YOUR/XXX.sql", "K1=V1,K2=V2,...")

# 其他操作
...

到了这里，关于解决方案：如何在 Amazon EMR Serverless 上执行纯 SQL 文件？的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！