一种流处理和批处理的切换方法及切换装置与流程

1.本发明属于金融数据处理的技术领域，尤其涉及一种流处理和批处理的切换方法及切换装置。


背景技术：

2.在金融数据处理领域，流处理与批处理是两种不同的计算模式，通常需要使用不同的工具和技术进行处理。然而，随着实时性需求的增加，业务系统往往需要能够同时处理实时流数据和历史批量数据。
3.flink（apache flink）是一种分布式流处理和批处理计算引擎，它将流处理和批处理能力融合在一个统一的框架中。flink流批一体化的原理是通过将数据流划分为有界流（bounded stream）和无界流（unbounded stream）来实现。flink流批一体的特点在于其提供了一致的编程模型和api接口，可以无缝地在流处理和离线批处理之间进行切换。这使得开发人员和数据工程师可以共享同一套代码和逻辑，以便同时满足实时处理和批量分析的需求。这种灵活性使得用户能够在一个集群上进行实时流处理和离线批处理任务，从而降低了管理成本并提高了计算效率。
4.然而，传统flink流批一体技术针对流处理和批处理的切换逻辑，仅仅是基于交易时段和闭市时段切换流处理和批处理，但在交易时段时，若遇到突发数据流，由于流处理以实时连续的方式处理数据流，数据按照到达顺序逐条处理。故在遇到突发数据流时，流处理常常出现过载情况，进而导致服务器宕机，这是一个亟需解决的技术问题。


技术实现要素：

5.有鉴于此，本发明实施例提供了一种流处理和批处理的切换方法、切换装置、终端设备以及计算机可读存储介质，以解决在遇到突发数据流时，流处理常常出现过载情况，进而导致服务器宕机的技术问题。
6.本发明实施例的第一方面提供了一种流处理和批处理的切换方法，所述切换方法包括：基于flink切换流处理模式和批处理模式；若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；
根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。
7.进一步地，所述根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分的步骤，包括：将多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率代入如下公式一，得到所述流处理效益评分；所述公式一为：
8.其中，表示所述匹配评分，表示第i个所述采样时间内的数据输入量，表示所述最大数据输入量，表示素数平均数据输入量，k表示所述采样时间的采样数量，表示第j个处理请求对应的所述平均处理时间，表示第j个处理请求对应的所述平均等待时间，n表示所述处理请求的数量，l表示所述cpu平均使用率，r表示所述内存平均占用率。
9.进一步地，所述根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略的步骤，包括：将所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，构建为当前状态向量；获取预存的多个聚类中心向量以及多个所述聚类中心向量各自对应的权重系数；所述多个聚类中心向量是基于聚类算法，在多个历史状态向量中计算而得；分别计算所述当前状态向量与多个所述聚类中心向量之间的各自对应的向量距离；将多个所述聚类中心向量各自对应的所述向量距离和所述权重系数进行加权求和，得到所述目标参数；根据所述目标参数所处的数值范围，匹配所述数值范围对应的批处理控制策略。
10.进一步地，在所述获取预存的多个聚类中心向量以及多个所述聚类中心向量各自对应的权重系数的步骤之前，还包括：获取多个历史样本数据对应的所述历史状态向量，计算多个所述历史状态向量之间的相似度，构建由多个所述相似度组成的相似度矩阵；根据所述相似度矩阵计算变换矩阵，并根据所述变换矩阵和所述相似度矩阵计算归一化目标矩阵；
基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个初始中心向量；基于所述初始中心向量和所述多个样本用户特征向量进行聚类计算，得到多个所述聚类中心向量。
11.进一步地，在所述基于所述初始中心向量和所述多个样本用户特征向量进行聚类计算，得到多个所述聚类中心向量的步骤之后，还包括：获取多个所述聚类中心向量各自对应的先验概率；将所述聚类中心向量和所述先验概率代入公式二，得到多个所述聚类中心向量各自对应的权重系数；所述公式二为：
12.其中，表示第j个聚类中心的权重系数，表示第i个历史状态向量在第j个聚类中心的高斯分布，表示第j个聚类中心的均值向量，表示第j个聚类中心的协方差矩阵，k表示聚类中心的数量，表示第j个聚类中心的先验概率，表示第k个聚类中心的先验概率，a表示历史状态向量的数量。
13.进一步地，所述基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个初始中心向量的步骤，包括：基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个特征值以及所述多个特征值各自对应的特征向量；按照数值大小，将前k个特征值对应的特征向量作为所述初始中心向量。
14.进一步地，所述根据所述相似度矩阵计算变换矩阵，并根据所述变换矩阵和所述相似度矩阵计算归一化目标矩阵的步骤，包括：将所述相似度矩阵代入如下公式三，得到所述变换矩阵；所述公式三为：
15.其中，表示所述变换矩阵，表示所述相似度矩阵中第i行第j列的数值；将所述变换矩阵与所述相似度矩阵相减，得到未归一化目标矩阵；将所述未归一化目标矩阵与所述变换矩阵相除，得到所述归一化目标矩阵。
16.本发明实施例的第二方面提供了一种流处理和批处理的切换装置，包括：第一切换单元，用于基于flink切换流处理模式和批处理模式；第一获取单元，用于若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；第一计算单元，用于根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；第二获取单元，用于若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；
第二计算单元，用于根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；第二切换单元，用于将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。
17.本发明实施例的第三方面提供了一种终端设备，包括存储器、处理器以及存储在所述存储器中并可在所述处理器上运行的计算机程序，所述处理器执行所述计算机程序时实现上述第一方面所述方法的步骤。
18.本发明实施例的第四方面提供了一种计算机可读存储介质，所述计算机可读存储介质存储有计算机程序，所述计算机程序被处理器执行时实现上述第一方面所述方法的步骤。
19.本发明实施例与现有技术相比存在的有益效果是：本发明通过基于flink切换流处理模式和批处理模式；若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。上述方案，先根据平均处理时间、平均等待时间、数据输入量、cpu平均使用率以及内存平均占用率，计算流处理效益评分。进而根据处理效益评分确定是否将流处理模式切换至批处理模式。在确定切换至批处理模式，则根据流处理效益评分、激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量，匹配批处理控制策略，以实现批处理的切换。通过上述切换逻辑，将流处理切换至批处理，避免了遇到突发数据流时，流处理常常出现过载情况。其中，批处理控制策略定义了数据块长度和分布式处理数量，通过适宜的数据块长度和分布式处理数量，提高了批处理效率。
附图说明
20.为了更清楚地说明本发明实施例中的技术方案，下面将对实施例或相关技术描述中所需要使用的附图作简单地介绍，显而易见地，下面描述中的附图仅仅是本发明的一些实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其它的附图。
21.图1示出了本发明提供的一种流处理和批处理的切换方法的示意性流程图；图2示出了本发明提供的flink技术流处理模式切换流程图；
图3示出了本发明一实施例提供的一种流处理和批处理的切换装置的示意图；图4示出了本发明一实施例提供的一种终端设备的示意图。
具体实施方式
22.以下描述中，为了说明而不是为了限定，提出了诸如特定系统结构、技术之类的具体细节，以便透彻理解本发明实施例。然而，本领域的技术人员应当清楚，在没有这些具体细节的其它实施例中也可以实现本发明。在其它情况中，省略对众所周知的系统、装置、电路以及方法的详细说明，以免不必要的细节妨碍本发明的描述。
23.本发明实施例提供了一种流处理和批处理的切换方法、切换装置、终端设备以及计算机可读存储介质，以解决在遇到突发数据流时，流处理常常出现过载情况，进而导致服务器宕机的技术问题。
24.首先，本发明提供了一种流处理和批处理的切换方法。请参见图1，图1示出了本发明提供的一种流处理和批处理的切换方法的示意性流程图。如图1所示，该流处理和批处理的切换方法可以包括如下步骤：步骤101：基于flink切换流处理模式和批处理模式；本技术是基于现有的flink技术的基础上进行自动切换，以克服在遇到突发数据流时，流处理常常出现过载情况。
25.请参见图2，图2示出了本发明提供的flink技术流处理模式切换流程图。如图2所示，flink技术通过cdc插件将监控源表中数据流的变化情况通过kafka的形式发送至处理器。处理器根据数据流的变化情况（分辨当前数据为实时数据或离线数据）。若数据流由离线数据变化至实时数据，则将批处理模式切换至流处理模式。
26.步骤102：若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；流处理模式以持续流动的数据为基础进行实时处理和分析。数据以流的形式逐条到达系统，并在短时间内被实时计算引擎（例如apache kafka、apache flink）进行处理。批处理模式以一定时间窗口内（如每天或每小时）累积的数据集合（批）为单位进行处理。数据在固定时间间隔内收集并存储，然后按照预定的时间执行作业或任务。由上述可知，流处理面对突发数据流容易出现过载情况，进而导致服务器宕机。而批处理模式为非实时数据处理，故面对突发数据流，不会出现过载情况。故本技术仅针对流处理模式进行自动切换，以避免在遇到突发数据流时，出现过载情况，进而导致服务器宕机的技术问题。
27.为了充分评估流处理模式的处理能力是否可以处理当前的输入数据，故需要获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率，以计算流处理效益评分（流处理效益评分用于表征流处理模式的当前处理能力）。
28.步骤103：根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；具体地，将多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平
均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率代入如下公式一，得到所述流处理效益评分；所述公式一为：
29.其中，表示所述匹配评分，表示第i个所述采样时间内的数据输入量，表示所述最大数据输入量，表示素数平均数据输入量，k表示所述采样时间的采样数量，表示第j个处理请求对应的所述平均处理时间，表示第j个处理请求对应的所述平均等待时间，n表示所述处理请求的数量，l表示所述cpu平均使用率，r表示所述内存平均占用率。
30.本技术综合考虑多方面的因素影响，且由于平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率对于流处理效益评分的计算均有一定的影响，故本技术基于平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率计算流处理效益评分，以实现对当前流处理模式的处理能力进行评估。其中，上述公式一是基于大量实验数据与验证而得，但不限于上述数学表达式。
31.步骤104：若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；当流处理增益评分小于第一阈值，则表示当前流处理模式的处理能力不足以处理当前数据流。故需要将流处理模式切换至批处理模式。其中，为了进一步提高批处理模式的处理效率，故本技术需要根据多个维度数据（用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量），计算批处理控制策略。批处理控制策略包括数据块长度和分布式处理数量等控制参量。
32.其中，激增指数是由用户输入的控制参数，是由用户根据人为经验确定的控制参数，使得批处理控制策略更符合实际环境需求。最小延迟参数是预设的参数值，是用于限制批处理模式的处理延迟。当前内存空间参数是指当前可以调用的内存空间资源。当前处理器算力参数是指当前可以调用的处理器算力资源。最大数据输入量表示多个处理请求中最大的数据量。最小数据输入量中最小的数据量。平均数据输入量是指多个处理请求中数据量的平均值。
33.步骤105：根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；具体地，步骤105具体包括步骤1051至步骤1055：步骤1051：将所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，构建为当前状态向量；步骤1052：获取预存的多个聚类中心向量以及多个所述聚类中心向量各自对应的
权重系数；所述多个聚类中心向量是基于聚类算法，在多个历史状态向量中计算而得；为了提高计算效率，故本技术在执行本技术方案之前预先计算多个聚类中心向量以及多个所述聚类中心向量各自对应的权重系数，具体计算逻辑如下：可选地，在步骤1052之前还包括步骤a至步骤f：步骤a：获取多个历史样本数据对应的所述历史状态向量，计算多个所述历史状态向量之间的相似度，构建由多个所述相似度组成的相似度矩阵。
34.在本技术中为了将噪声和局部异常点对数据的影响降到最小，故计算多个历史状态向量之间的相似度，构建由多个相似度组成的相似度矩阵。且相似度矩阵描述了数据点之间的相似性和差异性，反映了数据集的内在结构，以更好地实现聚类。且通过计算相似度矩阵，可以避免直接对原始数据进行计算，从而减少计算复杂度。同时，相似度矩阵的计算也可以方便地进行并行化操作，加快计算速度。
35.步骤b：根据所述相似度矩阵计算变换矩阵，并根据所述变换矩阵和所述相似度矩阵计算归一化目标矩阵。
36.为了进一步提高计算精度，故本技术基于相似度矩阵计算变换矩阵，变换矩阵中的数值可用于表征相似度矩阵中每个数值的重要性和相似性。且为了方便后续向量降维与聚类，故需要计算变换矩阵对应的归一化目标矩阵，具体计算过程如下：具体地，步骤b具体包括步骤b1至步骤b3：步骤b1：将所述相似度矩阵代入如下公式三，得到所述变换矩阵；所述公式三为：
37.其中，表示所述变换矩阵，表示所述相似度矩阵中第i行第j列的数值；步骤b2：将所述变换矩阵与所述相似度矩阵相减，得到未归一化目标矩阵；步骤b3：将所述未归一化目标矩阵与所述变换矩阵相除，得到所述归一化目标矩阵。
38.在本实施例中，通过相似度矩阵计算变换矩阵，该变换矩阵用于表征相似度矩阵中每个数值的重要性和相似性，提高了计算精度，并基于变换矩阵计算归一化目标矩阵，便于后续向量降维与聚类。
39.步骤c：基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个初始中心向量。
40.具体地，步骤c具体包括步骤c1和步骤c2：步骤c1：基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个特征值以及所述多个特征值各自对应的特征向量；步骤c2：按照数值大小，将前k个特征值对应的特征向量作为所述初始中心向量。
41.由于聚类算法需要选出k个初始聚类中心，以进行后续的聚类计算。故本技术通过主成分分析对所述归一化目标矩阵进行特征分解，得到多个特征值以及多个特征值各自对应的特征向量，并将前k个特征值对应的特征向量作为初始中心向量（即初始聚类中心）。其中，k的数值为超参数，即为预设的聚类中心数量。值得注意的是，由于主成分分析可以提取数据中的重要数据，故基于主成分分析得到的k个初始聚类中心具有比较强的代表性，进而提高了聚类算法的精度。
42.步骤d：基于所述初始中心向量和所述多个样本用户特征向量进行聚类计算，得到聚类中心向量。
43.可选地，在步骤d之前还包括步骤e至步骤f：步骤e：获取多个所述聚类中心向量各自对应的先验概率。
44.在聚类算法中，先验概率（prior probability）指的是在建模时对于每个簇的先验概率分布。它是指在没有任何数据观测之前，我们已经对每个簇的出现概率有一定的先验假设或知识。一般来说，先验概率可以通过领域专家的经验、历史数据分析等方法来确定。
45.步骤f：将所述聚类中心向量和所述先验概率代入公式二，得到多个所述聚类中心向量各自对应的权重系数；所述公式二为：
46.其中，表示第j个聚类中心的权重系数，表示第i个历史状态向量在第j个聚类中心的高斯分布，表示第j个聚类中心的均值向量，表示第j个聚类中心的协方差矩阵，k表示聚类中心的数量，表示第j个聚类中心的先验概率，表示第k个聚类中心的先验概率，a表示历史状态向量的数量。
47.本技术综合考虑多方面的因素影响，且由于权重系数、聚类中心的数量、聚类中心的高斯分布、聚类中心的均值向量、聚类中心的协方差矩阵和历史状态向量的数量对于聚类算法的计算精度均有一定的影响，故基于权重系数、聚类中心的数量、聚类中心的高斯分布、聚类中心的均值向量、聚类中心的协方差矩阵和历史状态向量的数量计算各个聚类中心的权重系数，以实现高精度聚类。其中，上述公式二是基于大量实验数据与验证而得，但不限于上述数学表达式。
48.步骤1053：分别计算所述当前状态向量与多个所述聚类中心向量之间的各自对应的向量距离；步骤1054：将多个所述聚类中心向量各自对应的所述向量距离和所述权重系数进行加权求和，得到所述目标参数；步骤1055：根据所述目标参数所处的数值范围，匹配所述数值范围对应的批处理控制策略。
49.不同的批处理控制策略与不同的数值范围存在预设的映射关系，故可基于数值范围匹配对应的批处理控制策略。批处理控制策略包括数据块长度和分布式处理数量。
50.其中，由于流处理模式处理的数据为无界数据，而批处理模式处理的数据为有界数据（即以数据块的形式进行数据处理）。故本技术通过定义批处理控制策略中的数据块长度，以控制有界数据的长度。可以理解的是，数据块长度过短时，无法充分利用当前的算力资源，导致计算效率低下。而数据块长度过长时，需要较长的处理时间，增大了处理延时。故需要设定合理的数据块长度，以提高计算效率，减小延时。
51.而分布式处理通过对待处理数据进行并行处理，为了分配合理的分布式资源，故通过分布式处理数量，限制并行处理的数量（即分布式处理数量是指并行处理的数量）。
52.其中，分布式批处理是一种利用多台计算机协同完成数据处理的技术，通常涉及到数据的分片、分发、并行执行和结果汇总等步骤。可基于如下三种分布式方案进行并行处理：第一种分布式方案：基于 mapreduce 模型的方案：mapreduce 是一种著名的分布式编程模型，由google 提出，用于处理大规模数据集。mapreduce 模型将数据处理分为两个阶段：map和reduce。map阶段负责将输入数据切分为多个键值对，并对每个键值对进行映射操作，生成中间结果。reduce阶段负责将具有相同键的中间结果进行归约操作，生成最终结果。mapreduce模型可以利用分布式文件系统（如 hdfs）和分布式调度系统（如 yarn）来实现数据的存储和任务的调度。mapreduce模型的优点是简单易用，可以自动处理数据切分、任务分配、容错等问题。mapreduce模型的缺点是不够灵活，只适合一些简单的批处理场景，对于复杂的数据处理流程或需要多次迭代的算法，效率较低。基于mapreduce模型的典型框架有hadoop mapreduce1 和apache spark2。
53.第二种分布式方案：基于dag（有向无环图）模型的方案：dag是一种表示数据处理流程中各个任务之间依赖关系的图结构。dag模型将数据处理流程抽象为一个由多个顶点和边组成的有向无环图，其中顶点表示任务，边表示任务之间的数据流或控制流。dag模型可以利用dag引擎来解析和执行dag 图，根据任务之间的依赖关系来确定任务的执行顺序和并行度，并根据资源情况来动态调整任务的分配和调度。dag模型的优点是灵活高效，可以支持复杂的数据处理流程和多次迭代的算法，同时可以实现精细化的资源管理和容错机制。dag模型的缺点是编程难度较高，需要用户自己定义dag图，并关注dag 引擎的内部实现细节。基于dag模型的典型框架有apache flink3和apache storm。
54.第三种分布式方案：基于任务调度框架的方案：任务调度框架是一种基于现有业务应用服务集群构建分布式批处理逻辑的方案，通常不需要额外部署专门的数据平台或引擎。任务调度框架通常提供了一些通用的功能和接口，如定时触发、任务定义、任务分片、任务执行、任务监控等，让用户只需要关注业务逻辑的实现，而不需要关心分布式协调的细节。任务调度框架的优点是轻量简便，可以快速集成到现有的业务系统中，同时可以利用业务集群的资源和能力来完成批处理任务。任务调度框架的缺点是功能较为有限，不太适合处理非常复杂或非常大规模的数据处理任务。基于任务调度框架的典型框架有 elasticjob和schedulerx。
55.步骤106：将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。
56.将流处理模式切换为批处理模式涉及以下步骤：
①
确定切换时间：确定切换流处理到批处理的时间点。
②
停止流数据输入：停止新的实时数据输入到流处理系统。这可以通过关闭数据源连接、暂停数据接收器或改变数据流方向等方式实现。
③
缓冲未处理数据：在停止数据输入后，需要确保所有尚未处理的流数据都被正确地缓存下来。可通过在内存或持久性存储中存储数据，或者将其写入临时文件来完成。
④
切换处理模式：根据系统设计，相应地更改流处理系统的配置或代码，以适应批处理。
57.作为本发明的一个可选实施例，若在切换至批处理模式后，检测到输入数据流低于阈值，则可将批处理模式切换至流处理模式。
58.在本实施例中，通过基于flink切换流处理模式和批处理模式；若当前的处理模式
为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。上述方案，先根据平均处理时间、平均等待时间、数据输入量、cpu平均使用率以及内存平均占用率，计算流处理效益评分。进而根据处理效益评分确定是否将流处理模式切换至批处理模式。在确定切换至批处理模式，则根据流处理效益评分、激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量，匹配批处理控制策略，以实现批处理的切换。通过上述切换逻辑，将流处理切换至批处理，避免了遇到突发数据流时，流处理常常出现过载情况。其中，批处理控制策略定义了数据块长度和分布式处理数量，通过适宜的数据块长度和分布式处理数量，提高了批处理效率。
59.如图3本发明提供了一种流处理和批处理的切换装置3，请参见图3，图3示出了本发明提供的一种流处理和批处理的切换装置的示意图，如图3所示一种流处理和批处理的切换装置包括：第一切换单元31，用于基于flink切换流处理模式和批处理模式；第一获取单元32，用于若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；第一计算单元33，用于根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；第二获取单元34，用于若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；第二计算单元35，用于根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；第二切换单元36，用于将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。
60.本发明提供的一种流处理和批处理的切换装置，通过响应于接收到的服务请求，识别所述服务请求的请求类型，并根据所述请求类型确定所述服务请求的处理难度指数；
获取多个待选服务器各自对应的选择概率以及资源数据，并根据所述选择概率和所述资源数据计算多个所述待选服务器各自对应的当前处理能力指数；所述选择概率是指待选服务器在预设次数内被选定为目标服务器的概率；根据所述处理难度指数和所述当前处理能力指数，计算多个所述待选服务器各自对应的处理效率；根据所述处理效率，在多个待选服务器中筛选目标服务器，并通过所述目标服务器处理所述服务请求。上述方案，通过服务请求类型确定服务请求的处理难度指数，通过待选服务器的选择概率以及资源数据确定当前处理能力指数，并根据处理难度指数和当前处理能力指数确定待选服务器各自的处理效率，以根据处理效率选择目标服务器。由于本发明基于多个维度（请求类型、选择概率以及资源数据）综合判断待选服务器的处理效率，使得均衡负载更加贴合实际情况，故可以很好地避免服务器出现过载等情况。
61.图4是本发明一实施例提供的一种终端设备的示意图。如图4所示，该实施例的一种终端设备4包括：处理器40、存储器41以及存储在所述存储器41中并可在所述处理器40上运行的计算机程序42，例如一种流处理和批处理的切换的程序。所述处理器40执行所述计算机程序42时实现上述各个一种流处理和批处理的切换方法实施例中的步骤，例如图1所示的步骤101至步骤1046。或者，所述处理器40执行所述计算机程序42时实现上述各装置实施例中各单元的功能，例如图3所示单元31至36的功能。
62.示例性的，所述计算机程序42可以被分割成一个或多个单元，所述一个或者多个单元被存储在所述存储器41中，并由所述处理器40执行，以完成本发明。所述一个或多个单元可以是能够完成特定功能的一系列计算机程序指令段，该指令段用于描述所述计算机程序42在所述一种终端设备4中的执行过程。例如，所述计算机程序42可以被分割成各单元的具体功能如下：第一切换单元，用于基于flink切换流处理模式和批处理模式；第一获取单元，用于若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；第一计算单元，用于根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；第二获取单元，用于若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；第二计算单元，用于根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；第二切换单元，用于将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。
63.所述终端设备中包括但不限于处理器40和存储器41。本领域技术人员可以理解，图4仅仅是一种终端设备4的示例，并不构成对一种终端设备4的限定，可以包括比图示更多
或更少的部件，或者组合某些部件，或者不同的部件，例如所述一种终端设备还可以包括输入输出设备、网络接入设备、总线等。
64.所述处理器40可以是中央处理单元(central processing unit，cpu)，还可以是其他通用处理器、数字信号处理器 (digital signal processor，dsp)、专用集成电路 (application specific integrated circuit，asic)、现成可编程门阵列 (field-programmable gate array，fpga) 或者其他可编程逻辑器件、分立门或者晶体管逻辑器件、分立硬件组件等。通用处理器可以是微处理器或者该处理器也可以是任何常规的处理器等。
65.所述存储器41可以是所述一种终端设备4的内部存储单元，例如一种终端设备4的硬盘或内存。所述存储器41也可以是所述一种终端设备4的外部存储设备，例如所述一种终端设备4上配备的插接式硬盘，智能存储卡（smart media card，smc），安全数字（secure digital，sd）卡，闪存卡（flash card）等。进一步地，所述存储器41还可以既包括所述一种终端设备4的内部存储单元也包括外部存储设备。所述存储器41用于存储所述计算机程序以及所述一种漫游控制设备所需的其他程序和数据。所述存储器41还可以用于暂时地存储已经输出或者将要输出的数据。
66.应理解，上述实施例中各步骤的序号的大小并不意味着执行顺序的先后，各过程的执行顺序应以其功能和内在逻辑确定，而不应对本发明实施例的实施过程构成任何限定。
67.需要说明的是，上述装置/单元之间的信息交互、执行过程等内容，由于与本发明方法实施例基于同一构思，其具体功能及带来的技术效果，具体可参见方法实施例部分，此处不再赘述。
68.所属领域的技术人员可以清楚地了解到，为了描述的方便和简洁，仅以上述各功能单元、模块的划分进行举例说明，实际应用中，可以根据需要而将上述功能分配由不同的功能单元、模块完成，既将所述装置的内部结构划分成不同的功能单元或模块，以完成以上描述的全部或者部分功能。实施例中的各功能单元、模块可以集成在一个处理单元中，也可以是各个单元单独物理存在，也可以两个或两个以上单元集成在一个单元中，上述集成的单元既可以采用硬件的形式实现，也可以采用软件功能单元的形式实现。另外，各功能单元、模块的具体名称也只是为了便于相互区分，并不用于限制本发明的保护范围。上述系统中单元、模块的具体工作过程，可以参考前述方法实施例中的对应过程，在此不再赘述。
69.本发明实施例还提供了一种计算机可读存储介质，所述计算机可读存储介质存储有计算机程序，所述计算机程序被处理器执行时实现可实现上述各个方法实施例中的步骤。
70.本发明实施例提供了一种计算机程序产品，当计算机程序产品在移动终端上运行时，使得移动终端执行时实现可实现上述各个方法实施例中的步骤。
71.所述集成的单元如果以软件功能单元的形式实现并作为独立的产品销售或使用时，可以存储在一个计算机可读取存储介质中。基于这样的理解，本发明实现上述实施例方法中的全部或部分流程，可以通过计算机程序来指令相关的硬件来完成，所述的计算机程序可存储于一计算机可读存储介质中，该计算机程序在被处理器执行时，可实现上述各个方法实施例的步骤。其中，所述计算机程序包括计算机程序代码，所述计算机程序代码可以
为源代码形式、对象代码形式、可执行文件或某些中间形式等。所述计算机可读介质至少可以包括：能够将计算机程序代码携带到拍照装置/终端设备的任何实体或装置、记录介质、计算机存储器、只读存储器（read-only memory，rom）、随机存取存储器（random access memory，ram）、电载波信号、电信信号以及软件分发介质。例如u盘、移动硬盘、磁碟或者光盘等。在某些司法管辖区，根据立法和专利实践，计算机可读介质不可以是电载波信号和电信信号。
72.在上述实施例中，对各个实施例的描述都各有侧重，某个实施例中没有详述或记载的部分，可以参见其它实施例的相关描述。
73.本领域普通技术人员可以意识到，结合本文中所公开的实施例描述的各示例的单元及算法步骤，能够以电子硬件、或者计算机软件和电子硬件的结合来实现。这些功能究竟以硬件还是软件方式来执行，取决于技术方案的特定应用和设计约束条件。专业技术人员可以对每个特定的应用来使用不同方法来实现所描述的功能，但是这种实现不应认为超出本发明的范围。
74.在本发明所提供的实施例中，应该理解到，所揭露的装置/网络设备和方法，可以通过其它的方式实现。例如，以上所描述的装置/网络设备实施例仅仅是示意性的，例如，所述模块或单元的划分，仅仅为一种逻辑功能划分，实际实现时可以有另外的划分方式，例如多个单元或组件可以结合或者可以集成到另一个系统，或一些特征可以忽略，或不执行。另一点，所显示或讨论的相互之间的耦合或直接耦合或通讯连接可以是通过一些接口，装置或单元的间接耦合或通讯连接，可以是电性，机械或其它的形式。
75.所述作为分离部件说明的单元可以是或者也可以不是物理上分开的，作为单元显示的部件可以是或者也可以不是物理单元，既可以位于一个地方，或者也可以分布到多个网络单元上。
76.应当理解，当在本发明说明书和所附权利要求书中使用时，术语“包括”指示所描述特征、整体、步骤、操作、元素和/或组件的存在，但并不排除一个或多个其它特征、整体、步骤、操作、元素、组件和/或其集合的存在或添加。
77.还应当理解，在本发明说明书和所附权利要求书中使用的术语“和/或”是指相关联列出的项中的一个或多个的任何组合以及所有可能组合，并且包括这些组合。
78.如在本发明说明书和所附权利要求书中所使用的那样，术语“如果”可以依据上下文被解释为“当...时”或“一旦”或“响应于确定”或“响应于监测到”。类似地，短语“如果确定”或“如果监测到[所描述条件或事件]”可以依据上下文被解释为意指“一旦确定”或“响应于确定”或“一旦监测到[所描述条件或事件]”或“响应于监测到[所描述条件或事件]”。
[0079]
另外，在本发明说明书和所附权利要求书的描述中，术语“第一”、“第二”、“第三”等仅用于区分描述，而不能理解为指示或暗示相对重要性。
[0080]
在本发明说明书中描述的参考“一个实施例”或“一些实施例”等意味着在本发明的一个或多个实施例中包括结合该实施例描述的特定特征、结构或特点。由此，在本说明书中的不同之处出现的语句“在一个实施例中”、“在一些实施例中”、“在其他一些实施例中”、“在另外一些实施例中”等不是必然都参考相同的实施例，而是意味着“一个或多个但不是所有的实施例”，除非是以其他方式另外特别强调。术语“包括”、“包含”、“具有”及它们的变形都意味着“包括但不限于”，除非是以其他方式另外特别强调。
[0081]
以上所述实施例仅用以说明本发明的技术方案，而非对其限制；尽管参照前述实施例对本发明进行了详细的说明，本领域的普通技术人员应当理解：其依然可以对前述各实施例所记载的技术方案进行修改，或者对其中部分技术特征进行等同替换；而这些修改或者替换，并不使相应技术方案的本质脱离本发明各实施例技术方案的精神和范围，均应包含在本发明的保护范围之内。

技术特征：
1.一种流处理和批处理的切换方法，其特征在于，所述切换方法包括：基于flink切换流处理模式和批处理模式；若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。2.如权利要求1所述的流处理和批处理的切换方法，其特征在于，所述根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分的步骤，包括：获取多个采样时间内所述数据输入量的最大数据输入量和多个采样时间内所述数据输入量的平均数据输入量；将多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率代入如下公式一，得到所述流处理效益评分；所述公式一为：其中， 表示所述匹配评分， 表示第i个所述采样时间内的数据输入量， 表示所述最大数据输入量， 表示素数平均数据输入量，k表示所述采样时间的采样数量， 表示第j个处理请求对应的所述平均处理时间， 表示第j个处理请求对应的所述平均等待时间，n表示所述处理请求的数量，l表示所述cpu平均使用率，r表示所述内存平均占用率。3.如权利要求1所述的流处理和批处理的切换方法，其特征在于，所述根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略的步骤，包括：将所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、
所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，构建为当前状态向量；获取预存的多个聚类中心向量以及多个所述聚类中心向量各自对应的权重系数；所述多个聚类中心向量是基于聚类算法，在多个历史状态向量中计算而得；分别计算所述当前状态向量与多个所述聚类中心向量之间的各自对应的向量距离；将多个所述聚类中心向量各自对应的所述向量距离和所述权重系数进行加权求和，得到所述目标参数；根据所述目标参数所处的数值范围，匹配所述数值范围对应的批处理控制策略。4.如权利要求3所述的流处理和批处理的切换方法，其特征在于，在所述获取预存的多个聚类中心向量以及多个所述聚类中心向量各自对应的权重系数的步骤之前，还包括：获取多个历史样本数据对应的所述历史状态向量，计算多个所述历史状态向量之间的相似度，构建由多个所述相似度组成的相似度矩阵；根据所述相似度矩阵计算变换矩阵，并根据所述变换矩阵和所述相似度矩阵计算归一化目标矩阵；基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个初始中心向量；基于所述初始中心向量和所述多个样本用户特征向量进行聚类计算，得到多个所述聚类中心向量。5.如权利要求4所述的流处理和批处理的切换方法，其特征在于，在所述基于所述初始中心向量和所述多个样本用户特征向量进行聚类计算，得到多个所述聚类中心向量的步骤之后，还包括：获取多个所述聚类中心向量各自对应的先验概率；将所述聚类中心向量和所述先验概率代入公式二，得到多个所述聚类中心向量各自对应的权重系数；所述公式二为：其中， 表示第j个聚类中心的权重系数， 表示第i个历史状态向量在第j个聚类中心的高斯分布， 表示第j个聚类中心的均值向量， 表示第j个聚类中心的协方差矩阵，k表示聚类中心的数量， 表示第j个聚类中心的先验概率， 表示第k个聚类中心的先验概率，a表示历史状态向量的数量。6.如权利要求4所述的流处理和批处理的切换方法，其特征在于，所述基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个初始中心向量的步骤，包括：基于主成分分析对所述归一化目标矩阵进行特征分解，得到多个特征值以及所述多个特征值各自对应的特征向量；按照数值大小，将前k个特征值对应的特征向量作为所述初始中心向量。7.如权利要求4所述的流处理和批处理的切换方法，其特征在于，所述根据所述相似度矩阵计算变换矩阵，并根据所述变换矩阵和所述相似度矩阵计算归一化目标矩阵的步骤，包括：
将所述相似度矩阵代入如下公式三，得到所述变换矩阵；所述公式三为：其中， 表示所述变换矩阵， 表示所述相似度矩阵中第i行第j列的数值；将所述变换矩阵与所述相似度矩阵相减，得到未归一化目标矩阵；将所述未归一化目标矩阵与所述变换矩阵相除，得到所述归一化目标矩阵。8.一种流处理和批处理的切换装置，其特征在于，所述切换装置包括：第一切换单元，用于基于flink切换流处理模式和批处理模式；第一获取单元，用于若当前的处理模式为流处理模式，则获取多个处理请求对应的平均处理时间、多个处理请求对应的平均等待时间、多个采样时间内的数据输入量、cpu平均使用率以及内存平均占用率；第一计算单元，用于根据多个处理请求对应的所述平均处理时间、多个处理请求对应的所述平均等待时间、多个采样时间内的所述数据输入量、所述cpu平均使用率以及所述内存平均占用率，计算流处理效益评分；第二获取单元，用于若所述流处理增益评分小于第一阈值，则获取由用户输入的激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量；第二计算单元，用于根据所述流处理效益评分、所述激增指数、所述最小延迟参数、所述当前内存空间参数、所述当前处理器算力参数、所述最大数据输入量、所述最小数据输入量和所述平均数据输入量，确定匹配批处理控制策略；所述批处理控制策略包括数据块长度和分布式处理数量；第二切换单元，用于将流处理模式切换为批处理模式，并基于所述批处理控制策略对待处理数据进行批处理。9.一种终端设备，包括存储器、处理器以及存储在所述存储器中并可在所述处理器上运行的计算机程序，其特征在于，所述处理器执行所述计算机程序时实现如权利要求1至7任一项所述方法的步骤。10.一种计算机可读存储介质，所述计算机可读存储介质存储有计算机程序，其特征在于，所述计算机程序被处理器执行时实现如权利要求1至7任一项所述方法的步骤。

技术总结
本发明适用于金融数据处理的技术领域，提供了一种流处理和批处理的切换方法及切换装置，所述切换方法包括：根据平均处理时间、平均等待时间、数据输入量、CPU平均使用率以及内存平均占用率，计算流处理效益评分。进而根据处理效益评分确定是否将流处理模式切换至批处理模式。在确定切换至批处理模式，则根据流处理效益评分、激增指数、最小延迟参数、当前内存空间参数、当前处理器算力参数、最大数据输入量、最小数据输入量和平均数据输入量，匹配批处理控制策略，以实现批处理的切换。通过上述切换逻辑，将流处理切换至批处理，避免了遇到突发数据流时，流处理常常出现过载情况。流处理常常出现过载情况。流处理常常出现过载情况。


技术研发人员：张前平 蔡仁光 王冉 张嘉枞
受保护的技术使用者：杭州迅杭科技有限公司
技术研发日：2023.08.31
技术公布日：2023/10/7