基于铁路数据服务平台的电务专业数据采集、共享及可视化研究

杨东盛，戚小玉，马小宁，李平，杨连报

（中国铁道科学研究院集团有限公司 电子计算技术研究所，北京 100081）

0 引言

电务系统对铁路系统的安全稳定运行至关重要。我国铁路运营情况复杂，电务系统规模不断扩大，设备功能越来越复杂，对保障电务系统长期安全、稳定运行的要求越来越迫切[1-2]。随着电务专业信息化建设的深化，新技术新设备大量投入，各设备子系统同步建设了相关检测、监测系统，产生和存储了海量数据，亟需运用大数据进行深入挖掘分析，优化安全生产流程，提升管理效率，更好地服务于铁路安全生产。

电务系统数据复杂，种类繁多，实时性要求较高[3]，采用传统接口方法要实现全专业数据的整合汇集并实现即时共享有较大困难。基于铁路数据服务平台[4]的采集和共享方案，利用Kafka分布式消息队列、RESTful接口、时序数据库、Flink数据流处理、数据包压缩等新技术，有效解决了采集共享方式不统一、时效性较低、网络带宽受限、大量数据共享延迟等问题，提升了电务各业务功能对数据服务平台中数据、资源、能力访问的稳定性、可靠性、时效性。同时，利用数据可视化技术将数据量大、维度复杂的特点通过多种方式进行可视化呈现，提高了数据使用效率，使业务人员能够直观地看到数据，便于发现问题和总结经验[5]。

1 电务专业数据采集共享流程

数据服务平台整体架构展示了数据由数据源经过数据服务平台采集、处理、存储、分析再共享的流程（见图1）。

图1 接口采集共享处理流程

在数据采集中，与传统数据采集直接将多数据源按不同接口方式接入数据库不同，新方案数据服务平台采用一种“数据总线”的方式实现数据源端的统一汇集，按需处理。具体方式为：无论数据采集方式、实时性要求如何，所有数据都先由数据源经相应的安全边界隔离后，接入并存储至采集端Kafka分布式消息队列[6-7]中，消息队列中暂存所有固定格式的序列化数据内容。随后，数据处理程序作为消息队列中数据的消费者，从“数据总线”中按需取出数据，再进行校验、存储或共享处理。

在这一采集共享流程中，采集端和共享端使用的Kafka分布式消息队列是一种可以提供高通量、低延迟、高可靠的数据传输服务，实时性响应性能满足现阶段电务专业用户的需求，同时，在采集层满足电务专业现有各类数据源系统的接口兼容性需求。共享端RESTful风格接口主要用于应用端和平台的数据交互，其拥有轻量化的设计，使用广泛且易于实现；FTP/SFTP服务则为普遍使用的文件传输协议，方便对数据量大但实时性需求不高的数据进行采集；Socket接口可以在网络中通过双向的通信连接实现数据交换，在此可为底层硬件设备传输数据提供转接。

采用此种数据采集方式的优势主要有：

（1）形成了平台外部数据源与内部数据处理的统一边界，可直接在消息队列中判断数据是否符合采集要求，无需再由数据库层面控制；

（2）数据通过“总线”方式存入，可实现并行处理，同一数据可以让共享程序、存储程序、分析程序同时消费，提高了数据的可共享性、可复用性和处理时效性；

（3）“总线”型架构结构灵活，易于扩展，未来如需在采集端增减接口，只需对Kafka队列进行适配，不会影响其他模块，改动量较小，同时也节省了多源数据对接的工作量；

（4）Kafka消息队列采用分布式架构，可以实现多节点的负载均衡，相对于单一接口方式单独处理，分布式架构对某一接口方式瞬时大数据量有较好的承载能力，提高了采集的稳定性和可靠性。

如图1所示，除数据源直接将数据送入Kafka的方式外，对于不支持Kafka方式的实时数据源，平台还提供使用Socket接口的数据集成模块进行数据接入；对于实时性要求不高的数据，平台提供FTP服务用于此类数据的采集。但无论使用何种接口方式，数据最终都会送入分布式消息队列中做进一步处理。

采集到数据后，平台除了使用Flink流处理服务实现大量高速的数据流处理，还在此过程中增加了采集数据的完整性校验，包括检查字段是否完整、类型是否一致、标识是否唯一等。增加的校验工作进一步保障了数据的真实可靠。相较于先存储再共享的低效率，为实现高速可靠的数据共享，平台将校验无误的数据分为3路同时处理，实时数据转换为JSON格式直接送入共享端Kafka队列共享，同时也存入时序数据库用于RESTful接口[8]的查询，第3路数据直接存入HIVE数据仓库用于大数据分析。

在数据共享中，平台也采用了类似采集端的“总线”机制，可为电务应用提供规范统一、格式标准的数据源。其中，实时数据直接送入共享端的Kafka消息队列，应用可直接消费Kafka中实时数据；或采用平台提供轮循调取的RESTful接口，支持相对实时的数据共享；对于非实时数据，平台统一使用RESTful接口实现。

此外，平台还支持大数据分析结果的共享使用。应用可以授权对HDFS/HIVE中数据进行分析预测处理，处理结果将采用非实时RESTful接口方式返回给应用。

2 实时数据采集与共享

2.1 实时数据采集共享方式

电务通信、信号相关业务数据具有实时性要求高、短时数据量大、可靠性需求高等特点。针对这些特点，对数据服务平台中的组件与服务进行了合理配置：

（1）对于实时性要求较高的需求，平台采用Kafka消息队列总线多线程处理，创新性地采用数据随进随出，独立线程同步存储的设计，极低延迟的实时数据流保障了共享端与应用接口的响应速度。

（2）对于短时数据量大的需求，平台采用分布式负载均衡配置，并创新使用高效的数据包压缩算法，降低带宽需求，还特别设计了针对特定数据简化内容中的冗余字段标识方法，进一步减小数据包大小。

（3）对于高可靠性保障需求，平台针对电务专业设计的Kafka分布式队列增加了副本数（利用备份策略实现即使集群中1台Kafka服务出现异常，其他服务也可以保障数据全量存储），从而保证数据冗余，防止数据丢失。同时，延长了Kafka消费数据的有效期，数据消费后不会立刻清除，从而保证了时间冗余，防止无据可查。此外，平台在流处理过程中增加了入库前的有效字段校验，确保采集共享全流程的高可靠性。

2.2 Kafka分布式消息队列

2.2.1 Kafka简介

Kafka消息队列是一个分布式流媒体平台，它是一种高吞吐量的分布式发布订阅消息系统。

Kafka主要有3种功能：一是可以发布和订阅消息流，分发订阅的消息；二是可以容错的方式记录消息流，并以文件的方式存储消息流；三是可以在消息发布的时候进行处理[9]。Kafka在系统或应用程序之间构建可靠的用于传输实时数据的管道，对于铁路数据服务平台形成了一个数据采集共享的总线，所有数据先通过各种方式汇集到Kafka，再通过Kafka分发到订阅它们的应用。

实际使用时，Kafka通过不同的主题（topic）对不同类型的数据进行分区处理。每个topic都被编入索引并存储时间戳。Kafka对外提供4种类型的接口：

（1）生产者接口。允许应用（数据源方）将数据流发布到1个或多个Kafka topic。

（2）消费者接口。允许应用（数据使用方）将数据流发布到1个或多个Kafka topic。

（3）数据流接口。将输入流转换为输出并生成结果。

（4）连接器接口。允许构建和运行可复用的生产者或消费者。

在Kafka中生产者将数据发布到某个topic中，消费者监听该topic并可靠地消费到数据。即生产者在topic中创建数据，消费者从这些topic中读取数据。为此，Kafka采用分布式设计，不同的topic由分区分隔，并在各节点之间复制。数据的消息体没有固定的格式，而是采用字节组的形式存储和传输。因此，可以利用它们来存储任何希望的格式对象，如JSON、XML等。

Kafka广泛使用在实时数据流水线的开发中，因为它可以提供高速高容量数据传输，这种高速数据通过Kafka的实时管道传递。发布的数据使用任何流平台或任何Kafka连接器进行订阅, 然后使用API 将订阅的数据推送到应用层使用。

2.2.2 Kafka传输的优势

Kafka作为一种分布式的发布订阅消息系统，其与JMS、RabbitMQ和AMQP等传统的消息代理不同，具有更高的吞吐量、可靠性和可扩展性，可以实现高速稳定的实时数据处理。主要是由于Kafka具有诸多优势：高通量（支持每秒数千条消息的吞吐量）、低延迟（能够以毫秒级的极低延迟处理消息）、高容错（分布式设计可以抵抗群集中的节点故障）、高并发性（允许以高并发读取和写入消息）、高耐久性（采用节点间消息复制机制，保证磁盘上数据或消息的持久性）、可扩展性（通过添加额外的节点不会导致任何停机的发生）、兼顾消息代理功能（可以替代传统的消息代理将来自发布者传递协议的消息转换为接收者传递协议的消息）等。

2.3 通信实时告警全流程Kafka传输

以通信专业为例，告警消息采用Kafka消息队列方式传输，具体处理流程见图2。

平台接收到告警消息后的处理流程如下：

（1）告警发生后，由通信数据源报送JSON格式的告警消息体到Kafka队列通信告警topic中（此时的消息体中包含告警ID、发生时间等基本信息，但由于告警刚刚产生，消息体中不包含派单、告警消除等状态信息，即在消息体中相应字段数据为空）；

图2 告警消息处理流程

（2）平台接收到告警消息后检查其完整性和唯一性（将告警ID作为唯一键，确认新接收告警是否为告警重复发送，并根据字段值判断是否满足格式要求，确保告警真实可靠）；

（3）平台将告警消息体实时上传到共享端Kafka队列，订阅告警消息的应用即刻收到告警消息；

（4）平台同步将告警消息体解析入库；

（5）平台将新增告警ID存入告警消息变更表（建立告警消息变更表的目的是方便无法使用Kafka的应用可以通过RESTful接口得知近期告警消息的新增和变更情况。告警消息变更表记录最新变更的告警ID和变更类型编码，应用查询该表新增内容后，可调用告警消息RESTful接口，获取变更告警的具体内容）。

告警状态变更的处理流程如下：

（1）告警状态发送变更后，通信数据源将变更消息以XML格式报送到与之前相同的topic中；

（2）平台接收后实时上传状态变更消息到共享端Kafka队列，应用即刻接收到变更消息；

（3）平台同步解析状态变更消息体，将告警确认或派单信息增加到之前插入的数据记录中；

（4）平台将变更告警ID和变更状态存入告警消息变更表。

告警消除的处理流程如下：

（1）告警清除后，由通信数据源报送JSON格式的告警清除消息体到相同的topic中；

（2）平台将告警清除消息体实时上传到共享端Kafka队列，应用即刻接收到清除信息；

（3）平台通过消息体中的告警清除ID在数据库中查找到先前的告警消息，修改记录中的清除状态，添加清除时间；

（4）平台将告警清除变更状态存入告警消息变更表。

2.4 RESTful接口轮循获取最新数据

对于可轮循调用的RESTful查询接口，应用可以连续调用该类接口获取最新数据。支持轮循调用的RESTful接口采用与非实时接口相同的调用方式，在使用上没有差异，其使用的数据同样来源于Kafka消息队列。对于调用量较高、较频繁的轮循接口，平台则可采取在独立硬件环境下单独部署的方式，以减少与普通非实时接口共同部署时的硬件资源消耗，从而避免影响普通接口的调用性能。

3 非实时数据采集与共享

3.1 采集与共享方式

针对电务专业非实时数据采集量较大、应用调用频繁、数据复用率高、具有长期累积分析价值等特点，平台设计了数据源定时上传FTP、平台监测文件并更新采集的方式。即对支持Kafka方式的数据源，依然采用与实时数据采集相同的序列化字节形式的数据包上传数据；对不支持Kafka的数据源，平台提供FTP前置服务器，数据源通过FTP服务定时上传CSV格式的数据文件，平台解析后将字符串形式的数据流送入消息队列中。CSV数据文件格式则根据具体数据表和业务需求的不同，由数据服务平台与数据源、应用三方商定。

对于应用层频繁调用，且同一数据可能多次重复调用的需求，数据服务平台提供了统一调用形式的RESTful接口实现。平台将非实时数据接口分为3类，一是与时间相关的时序数据接口，对于这类数据，应用可以通过以数据类型命名的接口，使用编码、开始时间、截止时间作为参数获取数据（此类数据采用时序存储方式，并使用时间字段作为数据表的索引，相对于关系型数据库，该设计不仅将存储空间减半，有效降低了I/O读写延时，查询速度也有极大提高）；二是与时间无关的静态数据接口，此类数据通常为履历和设备台账等基础信息，应用可以通过输入表名和SQL查询语句获取数据；三是对有特殊需求的应用功能，平台提供受限使用的通用SQL查询接口，该接口将SQL语句作为唯一参数查询数据库，因此可以实现多表关联等复杂的查询功能，为应用提供丰富、灵活的数据获取方式。

3.2 接口监测数据的FTP方式采集

以接口监测数据为例，数据通过FTP方式传输主要需要约定文件格式、命名方式、编码方式、数据内容、精度等信息。

（1）文件命名及编码方式。平台规定不同的数据源使用不同的FTP用户上传数据，并拥有独立的根目录用于上传。故文件命名不需要区分数据源，采用“数据分类标识-时间粒度-开始时间-结束时间-文件生成时间.csv” 的格式。

其中，时间格式为：YYYYMMDDHHMI（开始、结束时间）或YYYYMMDDHHMISS（文件生成时间）。例如，2019年1月2日15：02—15：17的信令PRI数据文件命名：PRI-15-201901021502-201901021517-20190102153449.csv。此外所有消息和文件的数据编码都采用UTF-8编码方式。

（2）文件内容。CSV文件内容的第1行为列表标题行；第2行以后为数据行，内容为标题行对应的数据值，字段间采用逗号分隔。其中，时间类字段的数据格式定义为“YYYY-MM-DD h24:mi:ss”。

（3）文件的传送和处理。各业务系统每时间周期按约定格式生成文件，并向FTP主动上传。每日产生的数据文件由平台负责压缩存储，并定期清除。

3.3 RESTful接口共享非实时数据

3.3.1 RESTful接口简介

数据服务平台非实时数据的共享统一由RESTful风格的接口提供[10]。其设计原则和条件主要包括：

（1）使用唯一的URL（统一资源定位符）来描述网络上的每一个实体；

（2）客户端通过一些HTTP协议中的动词，访问服务端的URL实现对服务器端资源的操作。表示操作方式的动词有GET、POST、PUT等。

RESTful接口常用于实现Web应用的前后端分离，适合于铁路数据服务平台向应用层共享数据。

3.3.2 RESTful接口调用

应用可通过访问平台接口URL实现对平台数据的访问。在调用平台提供的RESTful接口时，需遵循接口消息体格式、压缩方式等约定。

（1）消息体格式。平台接口的请求和响应均采用通用的JSON格式。

（2）消息体压缩。接口的请求体不做压缩，当响应消息体≥1 024 Byte时才会压缩（方便较多返回数据情况下响应消息体传输）。

（3）接口权限认证。用户按照相应的权限从数据服务平台获取数据，接口权限认证采用JWT（JSON Web Token）。

（4）返回数据格式。接口返回JSON数据包含status、message和data 3部分，status表示返回的状态码，message表示错误信息，data表示返回数据。

除了上述内容，为保障接口调用的可靠性，平台还采用如下约定：

（1）所有接口参数编码采用UTF-8编码方式；

（2）变量名和变量值不区分大小写（变量值是URL/URI、令牌或密码的除外），URL/URI区分大小写；

（3）接口中的GET和PUT操作需具备幂等性，也即执行多次操作和1次操作的影响一致；

（4）接口统一日期时间数据格式为：“YYYYMM-DD h24:mi:ss.SSS”。

4 电务数据可视化展示

铁路电务专业数据具有不同的实时性响应需求。例如，电务设备的运行情况与运输安全密切相关，设备告警信息尤其需要重点关注，既要关注实时告警，又要对非实时的历史告警信息进行关联性分析展示。此外，列车位置实时数据、问题库统计、施工、天窗、生产任务等非实时数据也是用户关注的内容。对上述数据进行可视化展示模型设计时，要具体分析各类接口的响应时间与数据的读取周期。同时，对电务专业数据可视化需求进行简单归类，按照关注点、使用人群、应用场景分类见表1。

表1 可视化需求分类

运维监测人员为及时发现问题，更关注实时数据，也就是运维监测类数据，包括告警监测、接口监测、服务运行状态监测等数据。分析调查人员比较关注非实时的历史数据，专注分析挖掘数据直接的关联关系，可通过逐级下钻的可视化方式满足这类用户的需求。指挥决策人员关心宏观指标与综合概览，可将上述实时与非实时数据可视化后的图表组合成驾驶舱页面，满足这类用户需求。

4.1 实时数据可视化展示

电务专业实时数据中，告警数据占较大比例，同时由于电务设备告警与运输安全关系紧密，这类数据也是用户最为关注的数据之一。在前端页面设置固定时间进行刷新的方式可对告警数据近似实时展示，为运维人员定位问题提供及时准确的参考（见图3、图4）。

图3 告警数据GIS定位及列表的近似实时可视化展示

图4 告警数据线路闪烁的近似实时可视化展示

利用地理信息平台中的铁路专业公用地理信息数据、铁路实景地理信息数据及铁路三维地理信息数据等多源空间信息数据，通过Web Service接口，将铁塔、基站、站房、直放站、信号机、应答器等电务设备的告警数据进行综合展现，为用户提供更加直观高效的展示。

4.2 非实时数据可视化展示

对于非实时的历史数据，主要从分析的角度，采用先宏观后微观的方式进行可视化展现。先用折线、柱图、饼图等基本组件进行宏观数据概况呈现，再通过下钻的方式对明细数据进行列表展示（见图5）。采用vue.js、three.js、svg.js等可视化图形库将电务专业数据、数据服务平台接口调用情况等数据可视化展示在前端页面，对运输生产及智能运维具有指导性意义。

图5 通信告警数据概览及下钻详单的可视化展示

5 结束语

基于铁路数据服务平台的电务数据实时采集与共享方案实现了电务数据的采集与共享。采用Kafka消息队列技术、RESTful接口技术等方法，实现了通信、信号相关专业数据的综合采集、存储，并提供标准化的共享接口，提供给电务应用统一使用，实现了高效率、高可靠性的标准化采集共享流程，为电务数据的集中、一体化展示提供完整的流程方案。同时，通过使用电务数据相关接口，提出电务数据的综合、集中、一体化的可视化展示方案，利用平台采集到的专业数据，实现综合、实时、丰富的可视化展示。