【最佳实践】使用Spark实时统计车辆启动时长

场景概述

本次实践帮助用户了解CS服务的基本功能，利用CS服务的Spark自定义类型作业，对实时接入的车辆行驶数据做初步汇总，统计出每辆车单次启动的情况，输出车辆启动时间、车辆停止时间和车辆启动时长等结果。

本次实践的原始数据为车辆接入车联网的实时状态信息的简化模拟，一般车辆接入车联网的总线数据包括驾驶中上传的经纬度信息、速度、方向、acc状态、油门开合度、刹车开合度等信息，本实践中的模拟数据只抽取了车辆ID、acc状态、数据上报时间三个字段，格式为csv。

本次实践通过Spark Streaming中状态管理的能力，根据车辆实时接入的数据，统计出每辆车单次启动的时长，并将结果输出。相比先将原始数据写入磁盘，再进行离线分析的方案来说，采用流计算的方案更实时，也能减少磁盘写入，从而节省成本。

本次实践中，为了简化流程，流计算的source和sink均为数据接入服务（DIS）。CS服务支持丰富的source和sink类型，具体请参见《实时流计算服务流生态开发指南》。

本次实践基本流程如下所示：

1. 准备Spark样例程序
2. 准备Spark样例数据
3. 创建集群
4. 创建作业
5. 查看作业执行情况和结果

准备Spark样例程序

1. 创建OBS桶，用于存放CS样例程序、日志，以及输出结果转储。
   1. 登录“公有云管理控制台”。
   2. 在“服务列表”中，选择“存储 > 对象存储服务”。
   3. 单击“创建桶”，创建一个名称为“cs-test”的桶。存储类别和桶策略分别保持默认值。
   4. 单击“立即创建”，完成OBS桶的创建。

2. 下载样例程序“SparkStreamingCountOnTime-assembly-1.0.jar”至本地。
   1. 在https://cloudstreamservice.obs.cn-north-1.myhuaweicloud.com/cs-sparkstreaming-CarAccTime.zip获取“cs-sparkstreaming-CarAccTime.zip”压缩包。
   2. 在您的工作目录解压。

3. 进入“cs-test”桶，单击“上传文件”，将“cs-sparkstreaming-CarAccTime”目录下的“SparkStreamingCountOnTime-assembly-1.0.jar”上传。

准备Spark样例数据

1. 创建DIS通道，用于接收车辆实时数据，即Spark样例数据，供CS作业读取。
   1. 登录“公有云管理控制台”。
   2. 在“服务列表”中，选择“EI企业智能 > 数据接入服务”。
   3. 单击“购买接入通道”，进入“购买接入通道”页面。
   4. 配置通道信息如下。

      区域：华北-北京一

      通道名称：sparkstreaming-input

      通道类型：普通

      分区数量：1

      生命周期（天）： 1

      源数据类型：BLOB

      图1 创建用于接收车辆实时数据的DIS通道
      
   5. 单击“立即购买”，完成DIS通道的创建。

2. 按照和1同样的方法，创建一个名称为“sparkstreaming-output”的通道，用于接收CS作业的输出数据。
3. 在IAM创建DIS委托。

   用户创建DIS通道，选择将数据转储到OBS中，需要通过创建IAM委托授权DIS服务访问用户的OBS资源。

   1. 登录“公有云管理控制台”。
   2. 在“服务列表”中，选择“管理与部署 > 统一身份认证服务”。
   3. 单击“委托”，选择“创建委托”。
   4. 参见图2配置DIS委托信息，创建一个名称为“test”的DIS委托。
      图2 配置DIS委托信息
      
   5. 单击“确定”，完成DIS委托创建。

4. 配置DIS通道的数据转储任务。
   1. 登录“公有云管理控制台”。
   2. 在“服务列表”中，选择“EI企业智能 > 数据接入服务”。
   3. 在左侧导航栏中单击“通道管理”，单击名称为“sparkstreaming-output”的通道，进入通道详情页面。
   4. 单击“转储管理”页签，再单击“添加转储任务”。
   5. 参见图3配置转储任务道信息后，单击“立即创建”。
   说明：

   • 数据转储地址：选择在1中创建的名称为“cs-test”的桶。
   • IAM委托：请选择在3中创建的DIS委托名称“test”。
   图3 配置转储任务信息
   

5. 下载样例程序“SparkStreamingCountOnTime-assembly-1.0.jar”至本地。
   1. 在https://cloudstreamservice.obs.cn-north-1.myhuaweicloud.com/cs-sparkstreaming-CarAccTime.zip获取“cs-sparkstreaming-CarAccTime.zip”压缩包。
   2. 在您的工作目录解压。

6. 运行应用程序包中的Jar包，生成模拟数据并发送到DIS通道。
   1. 在“cs-sparkstreaming-CarAccTime”目录下获取“SparkStreamingCountOnTime-assembly-1.0.jar”文件。
   2. 在“SparkStreamingCountOnTime-assembly-1.0.jar”文件所在的目录下执行如下命令。

      java -cp SparkStreamingCountOnTime-assembly-1.0.jar util.GenerateData CarNumber ReportPeriod DisEndpoint AK SK ProjectId Region DisChannel

      命令中参数说明如下表所示：

      表1 命令行参数配置

      参数名称	配置方法
      CarNumber	模拟数据中所包含的车辆的总数量，为整数值。建议设置为1000。
      ReportPeriod	模拟数据中每辆车上报数据的时间间隔，为整数值，单位秒。建议设置为10。
      DisEndpoint	DIS服务的endpoint地址。请参见地区和终端节点获取。例如：https://dis.cn-north-1.myhuaweicloud.com:443。
      AK/SK	请参见如下步骤获取： 登录“公有云管理控制台”。 鼠标移动至右上角用户名，在下拉菜单中选择“我的凭证”。 系统跳转至“我的凭证”页面，单击“管理访问密钥”页签。 单击“新增访问密钥”申请新密钥，按照提示输入密码与验证码，然后单击“确定”。 在弹出的“下载确认”提示框中，单击“确定”后，密钥会直接保存到浏览器默认的下载文件夹中。 打开下载下来的“credentials.csv”文件即可获取到访问密钥（AK和SK）
      ProjectId	请参见如下步骤获取： 登录“公有云管理控制台”。 鼠标移动至右上角用户名，在下拉菜单中选择“我的凭证”。 在“项目列表”页签，获取“华北-北京一”对应的项目ID。
      Region	区域名称，此处使用“cn-north-1”。
      DisChannel	DIS接收数据的通道，此处使用“sparkstreaming-input”。

   3. 命令行控制台上出现连续打印的“*** data sent”时，表示数据发送成功。

创建集群

1. 登录“公有云管理控制台”。
2. 在“服务列表”中，选择“EI企业智能 >实时流计算服务”，进入CS管理控制台。
3. 在左侧导航栏中，单击“集群管理”，进入“集群管理”页面。
4. 单击“创建集群”，参见表2配置集群基本信息。

   表2 集群配置信息

   参数名称	配置方式
   计费方式	使用按需计费方式。
   当前区域	单击管理控制台左上角的，选择“华北-北京1”。
   名称	“cs_demo”。
   描述	集群的相关描述，可为空。
   标签	如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在TMS中创建预定义标签，可为空。
   企业项目	在下拉列表中，选择用户已经在企业管理中创建的企业项目。 关于如何在企业管理中创建企业项目，请参见《企业管理用户指南》中“创建企业项目”章节。 如果您没有为集群选择企业项目，将使用缺省项目“default”。
   管理节点规格	独享集群使用的管理节点规格。与集群上运行作业的数量正相关。选择2 SPUs。
   集群扩容上限	集群为作业动态扩容的SPU上限数，默认100。集群创建后可修改。保持默认值。
   高级配置	可配置集群所属的VPC和子网网段，可根据网络规划调整。保持默认值。

   配置完成后，集群基本信息如下图所示。

   图4 创建集群
   

5. 单击“确认”，系统自动返回“集群管理”页面，可以在“状态”列中看到新建的集群正在“资源申请中”。

   新建集群的过程需要等待1~3分钟左右。当“状态”列显示“运行中”表示集群创建成功。

创建作业

1. 在CS管理控制台的左侧导航栏中，单击“作业管理”，进入“作业管理”页面。
2. 单击“新建作业”，弹出“新建作业”窗口。
3. 配置作业信息。
   • 类型：Spark自定义作业。
   • 名称：SparkStreamingCountOnTime。
   图5 新建作业
   

4. 单击“确认”，进入作业编辑页面。
5. 上传Jar包。
   图6 上传Jar包
   

   表3 上传Jar包参数配置

   参数名称	配置方法
   上传方式	选择“OBS上传”。 单击“从OBS选择文件”，选择在准备Spark样例程序中上传到“cs-test”桶中的“SparkStreamingCountOnTime-assembly-1.0.jar”文件。 单击“确认”按钮。
   Main Class	选择“指定”方式。 “类名”输入“car.CountOnTime”。 在“参数”框中输入如下对应参数。 DisEndpoint Region AK SK ProjectId InputStream Duration OutputStream 说明： DisEndpoint Region AK SK ProjectId的设置方法参考表1中对应参数。 InputStream为CS作业的输入通道，设置为“sparkstreaming-input”。 OutputStream为CS作业的输出通道，设置为“sparkstreaming-output”。 Duration为Spark Streaming程序的批次间隔，为整数值，单位秒，此处可设置为1。
   配置文件	无自定义配置文件，不设置。
   作业所属集群	选择在创建集群步骤中创建的“cs_demo”集群。

6. 单击左边“配置参数”，配置作业参数。
   图7 配置作业参数
   

   表4 作业参数配置

   参数名称	配置方法
   SPUs	作业使用的SPU总数，根据Driver、Executor的配置累计生成，不设置。
   Driver SPUs	Driver节点使用的SPU数量，此处使用默认值1。
   Executor个数	Executor节点的数量，此处使用默认值1。
   Executor SPUs	每个Executor节点使用的SPU数量，此处使用默认值1。
   保存作业日志	是否保存作业日志。 勾选开启作业日志保存功能。 单击“OBS桶”后的编辑框，在弹出的对话框中选择在1中创建的“cs-test”桶。 单击“OBS授权”进行授权。
   作业异常告警	开启此项功能可以在作业运行异常或者欠费的情况下发送短信、邮件等进行告警。此处暂不设置。
   异常自动重启	启动异常自动重启功能，则当作业异常时将自动重启并恢复作业。此处暂不设置。

7. 单击左边“选择运行集群”，“作业所属集群”选择在创建集群步骤中创建的“cs_demo”集群。
   图8 选择作业所属集群
   

8. 单击“提交”，进入作业配置确认页面，单击“确认”将作业提交并启动。

   提交作业后，系统将自动跳转到“作业管理”页面，新创建的作业将显示在作业列表中，在“状态”列中可以查看作业状态。作业提交成功后，状态将由“提交中”变为“运行中”。

查看作业执行情况和结果

1. 在CS管理控制台的左侧导航栏中，单击“作业管理”，进入“作业管理”页面。
2. 在“名称”一列中，单击运行中的“SparkStreamingCountOnTime”作业，进入“作业详情”页面。

3. 单击 “作业监控”页签，可以查看InputSize、SchedulingDelay、ProcessingTime、TotalDelay四个指标项。
   您也可以登录与集群相连的ECS节点访问更多监控信息。

   图9 查看作业监控
   

4. 单击 “运行日志”页签，可以查看作业的日志。
   单击右上角下拉框可以选择Driver或者Executor的日志，也可以选择StdOut或者StdError来查看。

   图10 查看作业日志
   

5. 在一个数据转储周期后（该数据转储周期在4中设置，默认为300秒），登录OBS管理控制台，进入“cs-test” 桶中，可以看到有“sparkstreaming-output”通道转储的文件生成，单击“下载”。
   图11 输出数据转储文件
   

6. 文件下载后打开，可以看到CS作业统计的车辆启动时长信息。
   格式为csv，字段含义分别为车辆ID、启动时间、停止时间、启动时长。

   图12 车辆启动时长统计信息