CN111753007A - A data aggregation system and aggregation method based on pluggable components under multi-system - Google Patents

Abstract

The invention discloses a pluggable component data aggregation system and an aggregation method based on multiple systems, wherein the method adapts a universal aggregation data method of multiple data sources, multiple protocols and multiple interface data sources through an aggregation collector of the data aggregation system, a target data source is aggregated through a hot-pluggable aggregation model device, the acquired data is converted into standard structured data through an intermediate data model conversion layer, a data stream is stored in a message middleware, and the data stream is called from the message middleware for analysis and processing. The invention does not need to collect the target data by modifying codes and adding plug-ins in the collected business system, has no perception to the collected business system, has no influence on the performance of the collection system, has no interference, is quick to deploy, has simple implementation conditions, and is compatible with various operating systems and various system platforms.

Description

A data aggregation system and aggregation method based on pluggable components under multi-system

technical field

The present invention relates to a data processing technology of multiple data sources under multiple protocols, how to adapt different data sources under multiple systems, and different structure aggregation methods, in particular to a data aggregation system and aggregation method based on pluggable components under multiple systems .

Background technique

At present, the adaptation of data sources in enterprise big data analysis applications or Internet data applications requires one-to-one customized development, which consumes a lot of time and workload. In order to solve the problem of multiple data sources in enterprise and Internet applications The following is how to automatically adapt the aggregation technology method of access operation data, especially the data collection and aggregation of data sources of multi-service systems. At present, customized development is usually used in the market to connect and adapt to different business systems. This method of development has a great impact on the manpower, material resources, and time of the data demander and data access party, and increases the time risk and technical risk of the data being connected to the system.

SUMMARY OF THE INVENTION

Purpose of the invention: Aiming at the problem of needing to aggregate business data of different protocols under multi-service systems in solving complex environments, the first object of the present invention is to provide a pluggable component data aggregation system under multi-system. Meanwhile, the second object of the present invention is to provide a method for data aggregation of pluggable components under multiple systems.

For achieving the above object, the technical scheme provided by the present invention is as follows:

A data acquisition system based on pluggable components under multiple systems, comprising a convergence server, a message middleware, a conversion processor and a consumption processor;

The aggregation server is deployed with pluggable aggregation components, which are used to adapt to different systems and data sources of different structures, collect the data streams of the business system data source systems through the aggregation components, and collect the collected data streams into the message middleware;

The message middleware is used to store the data stream collected by the aggregation server and the structured data processed by the conversion processor;

The conversion processor is a general matching data format processor based on the data source format, which processes the source data into structured data, and the format is standard json data;

The consumer processor, based on the structured data stream obtained from the message middleware, makes real-time call consumption through the consumer processor and streams the data to the big data real-time library.

Further, the aggregation server is a cloud server or a physical server, the cloud server or physical server and the collected business system network communicate with each other, the data port is open and shared, and the network protocol supports the mainstream TCP protocol.

Further, the aggregation server, message middleware and consumer processor are all compatible with server devices equipped with two or more operating systems, and realize data aggregation under two or more different operating systems. The operating systems include Windows operation. system, Unix operating system, linux operating system.

Further, the message middleware is kafka, and the configuration database is MySql and Oracle.

A method for data aggregation based on pluggable components under multiple systems for implementing the above aggregation system, the method includes the following steps:

(1) Convergence system establishment: open the data protocol interface of the business system, adapt the interface between the convergence server and the data flow of the connected business system, and convert the collected data flow through the conversion processor to form a structured data storage Go to the message middleware, and call the message middleware for message consumption processing;

(2) Data aggregation processing: The aggregation server collects the data stream of the collected business system through Filebeat. The data stream is filtered and processed by the data converter and then sent to the kafka or cluster in the message middleware for storage, and then to the message center through the Logstash tool. The data is obtained from the kafka in the file, passed to the Elasticsearch tool for data analysis and processing, and finally displayed to the kibana tool.

Further, in step (1), the aggregation server is adapted to the data of the collected business system, and the aggregated data is converted into structured data by the data conversion processor, and the structured data is stored in the message middleware. The specific methods are as follows:

Call data streams of different protocols of SocketServer, RestfulAPI, and JDBC based on Java language, and convert and encapsulate data of different structures into standard structured data objects through a unified structure processor, and re-encapsulate the data objects into unified structured json data. Tag the data and store the data objects in the message middleware.

Further, when the consumer processor obtains data from the message middleware, it can process the data objects in a multi-process and multi-threaded manner. The message middleware adopts kafka, and the real-time database adopts ElasticSearch7.0 for data storage.

Further, the various types of data gathered by the method are subjected to data structure conversion through a data conversion processor to form a standard json data format, and the specific method of calling data from the message middleware for consumption processing is as follows:

First, obtain the data stream from the message middleware, split and reorganize the data packets through Flume or Logstash, and then save the data objects to Elastic Search.

Beneficial effect: Compared with the prior art, the remarkable effect of the present invention is as follows:

(1) The convergence system of the present invention can realize automatic adaptation of business data collection by deploying hot-swappable management convergence components on the convergence server, saving a lot of time and workload; at the same time, it supports compatible Windows, Centos, Unix, and Linux operations. The system is also compatible with 32-bit and 64-bit operating platforms, adapting to the convergence requirements of different business systems, more flexible and more convenient than the existing technology, with wide coverage, simplified technical access, and strong automatic adaptation effect;

(2) Based on the aggregation method provided by the aggregation system, the data aggregation server can be automatically adapted according to the business scenario, without requiring a lot of time, workload and technical research, and the R&D adaptation work to meet the data requirements. Access, data analysis, and data mining all provide very convenient conditions, and do not interfere with the aggregated business systems, do not require embedded integration coding, adopt a distributed architecture, exist independently in the form of micro-service components, and deploy quickly , the implementation conditions are simple, the expansion is supported, the new aggregator can be installed in the form of a plug-in, and the ability to uninstall the aggregator is also supported; on the other hand, the method adopts a multi-platform, multi-process and multi-thread architecture, which is compatible with Windows, Centos, Unix, and Linux operating systems are also compatible with 32-bit and 64-bit operating platforms, adapting to the data aggregation requirements of different business systems, providing a method for connecting multiple business systems and multiple data structures for enterprises, institutions and Internet enterprise applications.

Description of drawings

Fig. 1 is the main flow chart of the method of the present invention;

Fig. 2 is the concrete schematic flow chart of the method of the present invention;

FIG. 3 is a schematic diagram of the architecture of the hot-pluggable component aggregator according to the present invention;

FIG. 4 is a schematic diagram of a framework of an application system in Embodiment 2 of the present invention.

Detailed ways

In order to describe the technical solutions disclosed by the present invention in detail, the present invention is described in detail below with reference to the accompanying drawings.

The invention provides a data aggregation system and aggregation method based on pluggable components under multi-system. The method adapts to multi-data sources, multi-protocol and multi-interface data sources through the aggregation collector of the data aggregation system. A general-purpose data method, collects the target data source through a hot-pluggable aggregation modeler, collects the feedback data and converts it into standardized structured data through the intermediate data model conversion layer, and stores the data stream in the message middleware, Then call the data flow from the message middleware for analysis and processing. Specifically, by adapting the protocol interface of the business system, the collected data flow of the business system is aggregated, and the collected data flow is formed into structured data through the data conversion processor, and the structured data is stored in the message middleware, Then call the data from the message middleware for data consumption processing.

On the basis of the above technical solutions, this embodiment performs data source adaptation according to the collected business system requirements, and selects an adapter that matches the data source.

Example 1

As shown in Figure 1 and Figure 2, a method for data aggregation based on pluggable components under multiple systems provided by the present invention, the specific steps are as follows:

Firstly, the data of the collected business system is aggregated according to business requirements. First, it is adapted through the data of the collected end, and the data structure is adapted through the existing aggregator as the data adaptation condition. The collected business system data aggregator aggregates the business system data through the aggregator and is structured by the data converter. The data is converted into structured json data, and then the message middleware stores the structured data in the memory message middleware, and then the consumer processor calls the structured data for consumption processing.

Among them, the collection aggregator adopts microservices with microservice architecture, which specifically includes using Java language and its data processing methods to call data streams of different protocols such as SocketServer, restfulAPI, JDBC, etc., and convert and encapsulate data of different structures through a unified structure processor into Standard structured data objects, encapsulate the data objects into unified structured json data, label the data, and then store the data objects in the message middleware.

Correspondingly, a multi-system-based pluggable component data acquisition system constructed by implementing the above method includes the following parts:

The aggregation server is used to adapt to different data source structures, collect the operation data of the business system, and store the collected data packets in the message middleware.

The message middleware is used to store the data packets collected by the aggregation server into the message middleware, which stores the processed structured data.

Conversion processor: It is a general matching data format processor based on the data source format, which processes the source data into structured data and the format is json data.

The consumption processor is used to obtain structured data packets from the message middleware, consume them in real time through the consumption processor, and store the data packets in the big data real-time database, namely Elastic Search.

Wherein, the convergence server, message middleware and consumer processor are all compatible with various operating systems, including Windows operating system, Unix operating system and Linux operating system, and are compatible with various operating platforms.

Among them, the data packets of the business system are aggregated, and the collected data packets are structured and processed through the data conversion processor, and stored in the message middleware through the message intermediate, which includes: using Java technology to call different protocols such as SocketServer, restfulAPI, JDBC, etc. data flow, and convert the data packets of different structures through the unified structure processor, encapsulate them into standard structured data objects, and then encapsulate the data objects into unified structured json data, label the json data, and then convert the data objects into Store to message middleware. And it aggregates and accesses the data packets of the business system, supports concurrent aggregation of multiple data packets, manages the adapted aggregator through the hot-plug component manager, and organizes, coordinates, starts and manages the aggregator by the hot-pluggable component manager. Stop, and can support hot-plug management of components without stopping the application, support installation and uninstall management of new aggregators, and hot-plug aggregators only need to support the microservice architecture to meet the requirements. The aggregated data is converted into a standard json data format by the data conversion processor, and the consumption processing is invoked from the message middleware. Specifically, it includes: obtaining data packets from the message middleware, and processing the data through Flume or Logstash. The package is split and reorganized to save the data objects in Elastic Search. The schematic diagram of the hot-pluggable component aggregator architecture is shown in Figure 3.

A system based on the pluggable component data aggregation method under multiple systems formed by the present invention includes:

The aggregation server is used to adapt to different data source structures, collect the operation data of the business system, and store the collected data packets in the message middleware;

The message middleware is used to store the data packets collected by the aggregation server into the message middleware, which stores the processed structured data.

The consumer server is used to obtain structured data packets from the message middleware, consume the data in real time through the consumer processor, and store the data packets in the big data real-time database, namely Elastic Search.

In this embodiment, the collection server converts the aggregated business data with a data volume of 1,000 pieces/s, and the size of each data packet is 1KB, and then performs data conversion by the data conversion processor to form structured data, and the conversion speed is 1,000 pieces. /s, the size is 1KB as an example, the conversion time is less than 1 second. The conversion processor stores the data in the message middleware, and the data stored in the message middleware is consumed by the consumer processor. The data interaction is realized through Java communication, and the aggregation server is compatible with a variety of operating systems. And a variety of operating platforms, the message middleware, the server running on the consumer processor adopts Windows, Linux, Centos, Unix and other servers.

This example performs data adaptation for a single data source. If the adaptation is successful, this adapter is used, and if the adaptation fails, it exits. Compared with the prior art, the data adapter function is added, and the data self-adaptation capability is provided. At the same time, it adopts real-time data processing architecture, such as real-time middleware KAFKA, FLUME, Logstash, Elastic Search. It can ensure that the data needs to be processed within the effective time.

Example 2

Embodiment 2 is a schematic diagram of a system implemented based on the system architecture of the present invention, and a schematic diagram of its frame structure is shown in FIG. 4 . The aggregation system accesses the collected system. For example, the data source of the business system can be accessed by multiple groups, and then automatically adapts to the aggregator through the data adapter. For example, the source data of the business system is automatically adapted to the corresponding aggregator. Other data sources In the same way, the data packets gathered by the aggregator are converted through the intermediate data converter to form json data, and then the structured data processor converts the json data into structured data, and pushes the formatted data to the message middleware , and then the consumer processor communicates with the message middleware, consumes the data in the message middleware, and pushes the consumed data to Elastic Search for data storage.

Wherein, the business system data source is the data source data in the enterprise application, and the data adapter is an intermediate component that automatically adapts according to the data source; the hot-plug aggregator is the data aggregator corresponding to the data adapter; The data converter is the component that performs data processing and conversion for the data packets gathered by the aggregator; the structured processor is the necessary component for data to form structured data; the message middleware processor is the component that pushes the structured data to the message middleware; The message middleware is kafka; the database is MySql database, oracle database, etc.

All business systems to be aggregated are collectively referred to as business system data sources. All data transactions will be automatically adapted to the corresponding aggregator. The aggregation processor uses JAVA technology to process the collected data with the memory message middleware. Data transaction, and then the middleware kafka communicates with the consumer server, and finally the consumer server pushes the data to Elastic Search to complete the collection of the entire chained data.

In this embodiment, the convergence server is developed based on the JAVA language programming technology, and can also be developed based on other languages such as Python, Go, and the like. The extended aggregator should conform to the micro-service architecture, and the aggregator itself is a micro-service. In the present invention, a multi-process and multi-thread mode is adopted for concurrent processing, and the modules are independent and low-coupling.

The problem that this example cannot handle in the prior art is to achieve zero delay in data processing. The acquisition processor, message middleware, and consumer processor all have delay processing problems for data processing, and real-time data processing cannot be realized in real time. Consumption. The invention adopts an adaptable multi-data source aggregator, and the adaptable multi-data source adapter itself has the characteristics of innovation and practicability. The latency of the aggregation processor is ≤1 second, the latency of the middleware processor is ≤1 second, and the latency of the consumer processor is ≤1 second. The present invention does not need to acquire data by modifying codes and adding plug-ins in the collected business system, has no perception of the collected business system, has no performance impact on the collection system, has no interference, has quick deployment, simple implementation conditions, and is compatible with many operating systems and multiple operating platforms.

Claims (8)

1. A pluggable component data aggregation system based on multiple systems is characterized by comprising an aggregation server, a message middleware, a conversion processor and a consumption processor;

the convergence server is deployed with a pluggable convergence component and is used for adapting to data sources of different systems and different structures, collecting data streams of a service system data source system through the convergence component and collecting the collected data streams to a message middleware;

the message middleware is used for storing the data stream acquired by the convergence server and the structured data processed by the conversion processor;

the conversion processor is used for processing the source data into structured data based on a general matching data format processor of a data source format, wherein the format is standard json data;

and the consumption processor acquires the structured data stream from the message middleware, calls consumption in real time through the consumption processor and flows the data stream to the big data real-time library.

2. The multisystem-based pluggable component data aggregation system according to claim 1, wherein the aggregation server is a cloud server or a physical server, the cloud server or the physical server is in network intercommunication with the collected server system, a data port is open and shared, and a mainstream TCP protocol is supported in a network protocol.

3. The multisystem-based pluggable component data aggregation system according to claim 1, wherein the aggregation server, the message middleware, and the consumption processor are all capable of processing data sources from different operating systems, so as to realize data aggregation under two or more operating systems, and the operating systems include a Windows operating system, a Unix operating system, and a linux operating system.

4. The multi-system based pluggable component data aggregation system of claim 1, wherein the message middleware is kafka and the configuration database is MySql or Oracle.

5. A pluggable component data aggregation method based on multiple systems is characterized by comprising the following steps:

(1) the construction of a convergence system: opening a data protocol interface of a service system, carrying out interface adaptation on a convergence server and a data stream of the accessed service system, carrying out data conversion on the acquired data stream through a conversion processor to form structured data, storing the structured data into a message middleware, and calling the message middleware to carry out message consumption processing;

(2) data aggregation processing: the convergence server collects data flow of the collected service system through the Filebeat, the data flow is transmitted to kafka or a cluster in the message middleware for storage after being filtered by the data converter, data are acquired from the kafka in the message middleware through the Logstash tool and transmitted to the Elasticisarch tool for data analysis and processing, and finally the data are displayed through the kibana tool.

6. The method for converging data of pluggable components based on multiple systems according to claim 5, wherein the structured data formed in the step (1) is processed as follows:

calling different protocol data streams of a SocketServer, a restfulAPI and JDBC based on Java language, converting and packaging different structural data into a standard structured data object through a unified structure processor, packaging the data object into unified structured json data, labeling the data, and storing the data object to a message middleware.

7. The method for data aggregation based on pluggable components under multiple systems according to claim 5, wherein the message middle in step (1) is kafka, elastic search is used for data storage, and a data object packaged in the message middleware is called in a multi-process and multi-thread manner.

8. The method for data aggregation based on pluggable components under multiple systems according to claim 5, wherein the Logstash in step (2) includes splitting and reassembling the data stream when acquiring the message middleware data, and storing the processed data object in an Elastic Search.

Images