利用 Amazon EMR Serverless、Amazon Athena、Apache Dolphinscheduler 以及本地 TiDB 和 HDFS 在混合部署环境中构建无服务器数据仓库（二）Apache DolphinScheduler 集成以及 LOB 粒度资源消费分析

引言

在数据驱动的世界中，企业正在寻求可靠且高性能的解决方案来管理其不断增长的数据需求。本系列博客从一个重视数据安全和合规性的 B2C 金融科技客户的角度来讨论云上云下混合部署的情况下如何利用亚马逊云科技云原生服务、开源社区产品以及第三方工具构建无服务器数据仓库的解耦方法。

Apache DolphinScheduler 是一种与 EMR Serverless 解耦部署的多功能工作流调度程序，可确保高效可靠的数据编排和处理。对于金融科技客户，EMR Serverless 提供业务线（LOB）级别的精细资源消费分析，从而实现精确监控和成本优化。这一功能在金融领域尤其有价值。因为在该领域，运营敏捷性和成本效益至关重要。

本篇博客着重探讨 Apache DolphinScheduler 与 EMR Serverless 的集成以及 LOB 粒度的资源消费分析方案。

架构设计图

Apache DolphinScheduler 通常采用和 Hadoop 集群混合部署的方式部署。根据不同的调度工作负载的情况可以选择在 Hadoop 集群中 HDFS 的多台 Data Node 上进行部署。本博客探讨的数仓计算引擎 EMR Serverless 和 DolphinScheduler 是解耦部署的。在 3 个 EC2 实例上以集群模式部署 Apache DolphinScheduler 对 EMR Serverless 的 Job 进行编排。

DolphinScheduler 集群与其编排的 EMR 作业解耦部署，实现了整个系统的高可靠性：一个（EMR 作业或调度器）发生故障不会影响另一个（调度器或 EMR 作业）。

图 1：解决方案系统架构图

DolphinScheduler 集成和作业编排

Apache DolphinScheduler 是现代数据编排平台。以低代码敏捷创建高性能工作流程。它还提供了强大的用户界面，致力于解决数据管道中复杂的任务依赖关系，并提供开箱即用的各种类型的作业。Apache DolphinScheduler 由 WhaleOps 开发和维护，并以 WhaleStudio 的产品名称上架亚马逊云科技 Market place。

Apache DolphinScheduler 原生集成 Hadoop。从下面两点可以具体看出：第一，DolphinScheduler 集群模式默认建议部署在 Hadoop 集群上（通常在数据节点上）；第二，上传到 DolphinScheduler 资源管理器的 HQL 脚本默认存储在 HDFS 上，并且可以通过本机 hive shell 命令直接编排，如下所示：

Hive -f example.sql

此外，对于这个具体案例，编排 DAG 相当复杂，每个 DAG 包含 300 多个作业。几乎所有作业都是存储在资源管理器中的 HQL 脚本。

因此，只有成功完成下面列出的任务，才能实现 DolphinScheduler 和 EMR Serverless 之间的无缝集成。

步骤 1：将 DolphinScheduler 资源中心的存储层从 HDFS 切换到 S3

分别编辑文件夹 /home/dolphinscheduler/dolphinscheduler/api-server/conf 和文件夹 /home/dolphinscheduler/dolphinscheduler/worker-server/conf 下的 common.properties 文件。文件中需要修改的部分如下所示：

#resource storage type: HDFS, S3, OSS, NONE
#resource.storage.type=NONE
resource.storage.type=S3
# resource store on HDFS/S3 path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended
resource.storage.upload.base.path=/dolphinscheduler

# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=AKIA************
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=lAm8R2TQzt*************
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=us-east-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=<target bucket name>
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=s3.us-east-1.amazonaws.com

编辑并保存这两个文件后，通过在文件夹路径 /home/dolphinscheduler/dolphinscheduler/bin/ 下执行以下命令重新启动 api-server 和 worker-server。

bash ./binstart-all.sh
bash ./bin/stop-all.sh
bash ./bin/status-all.sh

存储层切换到 S3 是否成功可以通过 DolphinScheduler 资源中心控制台上传脚本来检查，然后检查是否可以在相关的 S3 桶文件夹中找到该文件。

步骤 2：确保通过 S3 直接上传的作业脚本可以通过 DolphinScheduler 资源中心控制台找到并操作

完成第一步，可以实现从 DolphinScheduler 资源中心控制台上传脚本，并且这些脚本存储在 S3 中。然而，在实战中，客户需要将所有脚本直接迁移到 S3。存储在 S3 中的脚本应通过 DolphinScheduler 资源中心控制台查找和操作。为了实现这一点，需要通过插入所有脚本的元数据来进一步修改资源中心名为“t_ds_resources”的元数据表。插入命令如下：

insert into t_ds_resources values(4, '<target_script_name>', 'wordcount.java','',1,0,2100,'2023-11-13 10:46:44', '2023-10-31 10:46:44', 2, '<target_script_name>',0);

步骤 3：让 DolphinScheduler DAG 编排器了解作业的状态（FAILED/SUCCESS/SCHEDULED/PENDING），以便 DAG 能够根据作业的具体状态前进或采取相关操作

如上所述，DolphinScheduler 已与 Hadoop 生态系统原生集成，HQL 脚本可以由 DolphinScheduler DAG 编排器通过 Hive -f xxx.sql 命令编排。因此，当脚本改为 shell 脚本或 python 脚本时（EMR 无服务器作业需要通过 shell 脚本或 python 脚本编排，而不是简单的 Hive 命令），DAG 编排器可以启动作业，但无法获取实时数据作业的状态，因此无法进一步执行工作流程。由于本例中的 DAG 非常复杂，因此修改 DAG 是不可行的，而是遵循直接迁移策略。

因此，编写以下脚本来实现作业状态捕获和处理。

Application ID 列表持久化

var=$(cat applicationlist.txt|grep appid1)
applicationId=${var#* }
echo $applicationId

通过 linux shell 启用 ds 步骤状态自动检查

app_state
{
  response2=$(aws emr-serverless get-application --application-id $applicationId)
  application=$(echo $response1 | jq -r '.application')
  state=$(echo $application | jq -r '.state')
  echo $state
}

job_state
{
  response4=$(aws emr-serverless get-job-run --application-id $applicationId --job-run-id $JOB_RUN_ID)
  jobRun=$(echo $response4 | jq -r '.jobRun')
  JOB_RUN_ID=$(echo $jobRun | jq -r '.jobRunId')
  JOB_STATE=$(echo $jobRun | jq -r '.state')
  echo $JOB_STATE
}

state=$(job_state)

while [ $state != "SUCCESS" ]; do
  case $state in
    RUNNING)
         state=$(job_state)
         ;;
    SCHEDULED)
         state=$(job_state)
         ;;
    PENDING)
         state=$(job_state)
         ;;
    FAILED)
         break
         ;;
   esac
done

if [ $state == "FAILED" ]
then
  false
else
  true
fi

DolphinScheduler 版本推荐

实战发现并不是最高版本的 DolphinScheduler 是最好的，DolphinScheduler 目前最高的版本是 3.2.1，使用落后几个版本会比较安全。本案例分别测试了 3.1.4、3.1.5、3.1.8，3.1.4 最稳定。

DolphinScheduler 安装指南

针对 DolphinScheduler 的部署安装已经有 blog 做了不错的总结。如 blog。这里不再赘述。

LOB 粒度资源消费分析

如前所述，企业客户，尤其是金融科技客户，有建立内部清算结算机制的需求。亚马逊云科技成本分配标记机制完美满足了这一要求。所有实例，无论是配置的还是无服务器的，都可以作为标签附加。可以通过 Web 控制台或亚马逊云科技的 CLI 将标签附加到实例。

标记后，您可以在亚马逊云科技账单/成本分配标签控制台中激活标签，如下图所示。

图 2 Cost Allocation Tags 在亚马逊云科技 Console 的显示示意

激活标签后，标签的状态立即更改为“Active”。需要注意的是，通过账单和成本管理/成本浏览器控制台可视化标签的财务数据几乎需要一天的时间。如图 3 所示，在右侧的 Tag 下拉框中选择 CostCenter 之后，中间的柱状图显示了打了 CostCenter 这个 Tag 的不同 Value 值的服务消费情况。这里，Value 的值设计成需要了解资源消费的 LOB 的名称即可实现在 LOB 粒度对资源消费情况进行统计以及可视化展现。

图 3 在 Billing 和 Cost Management Console 上按 Cost Center 的 Tag 显示资源消费情况

总结

DolphinScheduler 作为大数据作业调度工具在华人开发者中非常流行。然而，其原生部署环境在 hadoop 上的现状和亚马逊云科技持续创新的新一代 Serverless 架构的产品服务之间存在一些 gap。本文结合实战总结了填补这些 gap 的方法，并探讨了通过打 Tag 的方式实现 LOB 粒度资源消费数据统计及可视化的方法。

亚马逊AWS官方博客