Patent Description:
In industry, a typical stream processing system requires distributed messaging queues which provide a publish-subscribe model to make stream processing easy and flexible. The messaging queue has a dedicated service and storage that provides real-time data processing but introduces high resource costs.

<CIT> discloses a method whereby an indication of an input data stream comprising data records, stored at a stream management service, that are to be batched for a computation at a batch-oriented data processing service, is received. A set of data records of the input data stream are identified, based on respective sequence numbers associated with the records, for a particular iteration of the computation. Metadata associated with the particular iteration, comprising identification information associated with the set of records on which the computation is performed during the particular iteration, is saved in a repository.

<CIT> discloses a technology disclosed relating to managing resource allocation to task sequences in a stream processing framework. In particular, it relates to operating a computing grid that includes machine resources, with heterogeneous containers defined over whole machines and some containers including multiple machines. It also includes initially allocating multiple machines to a first container, initially allocating first set of stateful task sequences to the first container, running the first set of stateful task sequences as multiplexed units of work under control of a container-scheduler, where each unit of work for a first task sequence runs to completion on first machine resources in the first container, unless it overruns a time-out, before a next unit of work for a second task sequence runs multiplexed on the first machine resources. It further includes automatically modifying a number of machine resources and/or a number assigned task sequences to a container.

A stream processing system, which is built on top of a messaging queue, is able to provide real-time (sub-second or few second) data processing of data in a distributed messaging queue. While this may be advantageous in circumstances where real-time data processing is needed, providing this real-time data processing comes at a high resource cost. For example, the stream processing system, which is built on top of a messaging queue, requires a large amount of dedicated storage and processing resources to ensure that the new data is quickly processed. This is especially true in high data volume circumstances. In addition, it is typically required to know the number of partition files ahead of time so that the proper resources can be allocated. Accordingly, the stream processing system, which is built on top of a messaging queue, requires a company or the team that is implementing the stream processing system to have dedicated servers and storage for the system.

A batch processing system typically requires less resource cost than a stream processing system. For example, since the data in partition files is only processed after a given time window ends, fewer processing resources are needed because the data is processed in a large batch or chunk of data instead of event by event. Thus, the number of IOs can be reduced, leading to higher processing efficiency. However, since the processing is done after the end of the time window, there is high latency for receiving the processed results. For example, if the time window is a day, then the data will not be processed until after the day is done. Further, if any data from one time period is delayed in being provided to the system until after the end of the time period, that data will typically not be processed until the end of the next time window adding additional latency. An example of this is in the banking industry where a deposit received on a given day after <NUM> PM is added into the deposits for the next day. Thus, a batch processing system provides large data volume processing, but at a cost of delayed processing results.

Since both of the stream processing system and the batch processing system provide different advantages, it is common for a company or team to implement both systems. Thus, these companies or teams are required to have separate, dedicated resources for both systems. In addition, it quite common for the data that is to be processed to come from the same data producers and thus have to be duplicated so that both the stream processing system and the batch processing system are able to process the data in their respective ways. The need for two separate processing systems with their own dedicated resources can be costly.

The embodiments described herein aim to solve at least some of the above-mentioned problems by providing for a near-real-time stream processing system that is implemented using the same distributed file system as the batch processing system. In this way, data received by the distributed file system need not be duplicated. In some instances, the implementation of the near-real-time stream processing system may result in <NUM>% or more reduction in engineering and resource costs.

According to the present disclosure, a data container and one or more partition files within the data container are generated according to a defined partition window. The defined partition window specifies a first time range that controls when data is to be included in the one or more partition files of the data container, wherein the one or more partition files are generated during the partition window, the data received during the partition window being placed into the one or more partition files. After the end of the partition window, a batch processing system performs batch processing on the data in the one or more partition files. In addition a near-real-time stream processing system performs near real-time stream processing whereby: a) the data container is periodically scanned to determine if the one or more partition files are within a defined partition lifetime window, wherein he defined partition lifetime window specifies a second time range that controls how long the one or more partition files are considered active such that processing is to be performed on the one or more partitions, b) for each partition file within the defined lifetime window, one or more processing tasks are created based on an amount of data included in the one or more partition files, and c) tThe data in the one or more partition files is accessed and the one or more processing tasks are performed. The accessing of the data by the near-real-time processing system is from the same one or more partition files in the same container as the data accessed by the batch processing system to perform the batch processing. Information about the one or more partition files is recorded in a configuration data store.

In an embodiment, the one or more processing tasks comprise a sequential process where there is only one active processing task at any given time for a given partition file of the one or more partition files, a parallel process where two or more processing tasks are completed at substantially the same time by different processing instances for a given partition file of the one or more partition files, or a parallel process within a same processing instance where two or more processing tasks are assigned to be performed by the same processing instance for a given partition file of the one or more partition files.

In an embodiment, the one or more partition files are discovered dynamically at runtime. In another embodiment, the defined partition lifetime window is larger than the defined partition window.

In an embodiment, a second data container and one or more second partition files within the second data container are generated according to a second defined partition window. The second defined partition window specifies a third time range that has a beginning after an ending of the first defined partition window but before an ending of the defined partition lifetime window. Prior to the ending of the defined partition lifetime window, the one or more partition files in the first data container and the one or more second partition files in the second data container are scanned. The one or more partition files and the one or more second partition files are both considered active during the defined partition lifetime window One or more processing tasks are created for the one or more partition files and the one or more second partition files. The assigned processing tasks are performed. the information about the one or more partition files and information about the one or more second partition files are recorded in the configuration data store.

In an embodiment, the one or more partition files are closed when it is determined that the partition lifetime window has reached an end and it is determined that processing has been performed on all the data in the one or more partition files. In another embodiment, the information about the one or more partition files recorded in the configuration data store comprises information about the progress of the one or more processing tasks. In a further embodiment, the one or more processing tasks are defined by a partition file identifier, a start offset address location of the data, and an end offset address location of the data.

In an embodiment, a task manager component is configured to scan a data container to discover if the data container includes one or more partition files and upon discovery, to determine if the one or more partition files are within a defined partition lifetime window, the partition lifetime specifying a time range that controls how long the one or more partition files are considered active such that processing can be performed on the received data in the one or more partitions. The task manager component is further configured to create, for those partition files that are within the defined partition lifetime, one or more processing tasks. One or more task worker components are configured to receive the one or more processing tasks from the task manager component and perform processing on the data in the one or more partition files according to the one or more processing tasks. The task manager component is further configured to cause information about the one or more partition files and about progress of the one or more processing tasks to be stored in a configuration data store.

Embodiments disclosed herein are related to computing systems and methods for implementing a near-real-time stream processing system using the same distributed file system as a batch processing system. In one embodiment, a data container and one or more partition files within the data container are generated according to a defined partition window. The defined partition window specifies a first time range that controls when data is to be included in the one or more partition files of the data container. The data container is scanned to determine if the one or more partition files are within a defined partition lifetime window. The defined partition lifetime window specifies a second time range that controls how long the one or more partition files are considered active such that processing is to be performed on the one or more partitions. For each partition file within the defined lifetime window, one or more processing tasks are created based on an amount of data included in the one or more partition files. The data in the one or more partition files is accessed and the one or more processing tasks are performed. Information about the one or more partition files is recorded in a configuration data store.

Because the principles described herein may be performed in the context of a computing system, some introductory discussion of a computing system will be described with respect to <FIG>. Then, this description will return to the principles of the near-real-time stream processing system with respect to the remaining figures.

Computing systems may, for example, be handheld devices, appliances, laptop computers, desktop computers, mainframes, distributed computing systems, datacenters, or even devices that have not conventionally been considered a computing system, such as wearables (e.g., glasses). In this description and in the claims, the term "computing system" is defined broadly as including any device or system (or combination thereof) that includes at least one physical and tangible processor, and a physical and tangible memory capable of having thereon computer-executable instructions that may be executed by a processor.

As illustrated in <FIG>, in its most basic configuration, a computing system <NUM> typically includes at least one hardware processing unit <NUM> and memory <NUM>. The processing unit <NUM> may include a general purpose processor and may also include a field programmable gate array (FPGA), an application specific integrated circuit (ASIC), or any other specialized circuit. The memory <NUM> may be physical system memory, which may be volatile, non-volatile, or some combination of the two. The term "memory" may also be used herein to refer to non-volatile mass storage such as physical storage media. If the computing system is distributed, the processing, memory and/or storage capability may be distributed as well.

The computing system <NUM> also has thereon multiple structures often referred to as an "executable component". For instance, the memory <NUM> of the computing system <NUM> is illustrated as including executable component <NUM>. The term "executable component" is the name for a structure that is well understood to one of ordinary skill in the art in the field of computing as being a structure that can be software, hardware, or a combination thereof. For instance, when implemented in software, one of ordinary skill in the art would understand that the structure of an executable component may include software objects, routines, methods, and so forth, that may be executed on the computing system, whether such an executable component exists in the heap of a computing system, or whether the executable component exists on computer-readable storage media.

In such a case, one of ordinary skill in the art will recognize that the structure of the executable component exists on a computer-readable medium such that, when interpreted by one or more processors of a computing system (e.g., by a processor thread), the computing system is caused to perform a function. Such structure may be computer-readable directly by the processors (as is the case if the executable component were binary). Alternatively, the structure may be structured to be interpretable and/or compiled (whether in a single stage or in multiple stages) so as to generate such binary that is directly interpretable by the processors. Such an understanding of example structures of an executable component is well within the understanding of one of ordinary skill in the art of computing when using the term "executable component".

If such acts are implemented exclusively or near-exclusively in hardware, such as within a FPGA or an ASIC, the computer-executable instructions may be hard coded or hard wired logic gates.

While not all computing systems require a user interface, in some embodiments, the computing system <NUM> includes a user interface system <NUM> for use in interfacing with a user. The user interface system <NUM> may include output mechanisms 112A as well as input mechanisms 112B. The principles described herein are not limited to the precise output mechanisms 112A or input mechanisms 112B as such will depend on the nature of the device. However, output mechanisms 112A might include, for instance, speakers, displays, tactile output, holograms and so forth. Examples of input mechanisms 112B might include, for instance, microphones, touchscreens, holograms, cameras, keyboards, mouse of other pointer input, sensors of any type, and so forth.

Embodiments described herein may comprise or utilize a special purpose or general-purpose computing system including computer hardware, such as, for example, one or more processors and system memory, as discussed in greater detail below. Such computer-readable media can be any available media that can be accessed by a general purpose or special purpose computing system.

Computer-readable storage media includes RAM, ROM, EEPROM, CD-ROM or other optical disk storage, magnetic disk storage or other magnetic storage devices, or any other physical and tangible storage medium which can be used to store desired program code means in the form of computer-executable instructions or data structures and which can be accessed by a general purpose or special purpose computing system.

Transmissions media can include a network and/or data links which can be used to carry desired program code means in the form of computer-executable instructions or data structures and which can be accessed by a general purpose or special purpose computing system.

Computer-executable instructions comprise, for example, instructions and data which, when executed at a processor, cause a general purpose computing system, special purpose computing system, or special purpose processing device to perform a certain function or group of functions. Alternatively or in addition, the computer-executable instructions may configure the computing system to perform a certain function or group of functions.

The invention may also be practiced in distributed system environments where local and remote computing systems, which are linked (either by hardwired data links, wireless data links, or by a combination of hardwired and wireless data links) through a network, both perform tasks.

The remaining figures may discuss various computing system which may correspond to the computing system <NUM> previously described. The computing systems of the remaining figures include various components or functional blocks that may implement the various embodiments disclosed herein as will be explained. The various components or functional blocks may be implemented on a local computing system or may be implemented on a distributed computing system that includes elements resident in the cloud or that implement aspects of cloud computing. The various components or functional blocks may be implemented as software, hardware, or a combination of software and hardware. The computing systems of the remaining figures may include more or less than the components illustrated in the figures and some of the components may be combined as circumstances warrant. Although not necessarily illustrated, the various components of the computing systems may access and/or utilize a processor and memory, such as processor <NUM> and memory <NUM>, as needed to perform their various functions.

<FIG> illustrates a high-level view of a stream processing system <NUM>. As illustrated, the stream processing system <NUM> includes a data producer <NUM>, a data producer <NUM>, and any number of additional data producers as illustrated by the ellipses <NUM>. In operation, the various data producers, which may be service endpoints or client endpoints, generate data or events that are to be processed by the stream processing system. In some embodiments, the data producers are referred to as "producers".

The generated data is then provided to a distributed messaging queue <NUM>, where it is placed into various event files such as event file <NUM>, event file <NUM>, and any number of additional event files as illustrated by the ellipses <NUM> based on metadata such as an event key or other event file identifier. In some embodiments, the various event files are organized into a topic, such as topic <NUM>. The messaging queue may be implemented in a distributed file system so that there are multiple replicas of each of the event files.

The event files are processed by the stream processing system <NUM>, which may include multiple servers and processing instances. The stream processing system <NUM> uses a publisher/subscriber and related APIs associated with the messaging queue <NUM> to determine if any new data has been appended to the event files. If there has been new data appended to an event file, then the stream processing system will perform processing on this data. In this way, the data appended to the event files is generally processed in near-real-time. The processed data is then provided to a consumer <NUM>, a consumer <NUM>, or to any number of additional consumers as illustrated by the ellipses <NUM>, which in some embodiments subscribe to receive data related to different topics.

<FIG> illustrates a high-level view of a batch processing system <NUM>. As illustrated, the stream processing system <NUM> includes a data producer <NUM>, a data producer <NUM>, and any number of additional data producers as illustrated by the ellipses <NUM>. In operation, the various data producers, which may be service endpoints or client endpoints, generate data or events that are to be processed by the batch processing system. In some embodiments, the data producers <NUM>-<NUM> may be the same as the data producers <NUM>-<NUM> or may generate the same data or events as the data producers <NUM>-<NUM>. Thus, in such embodiments the data that is generated by the data producers is duplicated so that the data is provided to both the stream processing system <NUM> and the batch processing system <NUM>.

The generated data is then provided to a partition file system <NUM>. The partition file system <NUM> generates a logical data container such as data container <NUM> based on a given time window. For example, the data container <NUM> may be based on a time window that is one day or one hour. Within the data container <NUM>, the data received during the time window is placed into partition files such as partition file <NUM>, partition file <NUM>, and any number of additional partition files as illustrated at by ellipses <NUM>.

At the end of the given time window, the data in the partition files <NUM>-<NUM> is processed by the batch processing system <NUM>, which may include multiple servers and processing instances. The results may then be provided to a consumer <NUM> and any number of additional consumers as illustrated by the ellipses <NUM>.

Supposing that the time period for the data container <NUM> is one day, for example February <NUM>, <NUM>, then when it becomes February <NUM>, <NUM>, the partition file system <NUM> will generate a data container <NUM> for the time window of February <NUM>, <NUM>. Within the data container <NUM>, the data received during the time window of February <NUM>, <NUM> is placed into partition files such as partition file <NUM>, partition file <NUM>, and any number of additional partition files as illustrated at by ellipses <NUM>.

At the end of the February <NUM>, <NUM> time period, the data in the partition files <NUM>-<NUM> is processed by the batch processing system <NUM>. The results may then be provided to the consumers <NUM> and <NUM>. The ellipses <NUM> illustrate that any number of additional data containers for additional time windows may be generated by the partition file system <NUM> as needed.

As mentioned previously, the stream processing system <NUM> is able to provide real-time (sub-second or few second) data processing of the data in the distributed messaging queue <NUM>. While this may be advantageous in circumstances where real-time data processing is needed, providing this real-time data processing comes at a high resource cost. For example, the stream processing system requires a large amount of dedicated storage and processing resources to ensure that the new data appended to a partition file is quickly processed. This is especially true in high data volume circumstances. In addition, it is typically required to know the number of partition files so that the proper resources can be allocated. Accordingly, the stream processing system <NUM> requires a company or the like that is implementing the stream processing system to have dedicated servers and storage for the system.

The batch processing system <NUM> typically requires less resource cost. For example, since the data in the partition files is only processed after the time window ends, fewer processing resources are needed. In addition, less storage resources are typically required. However, since the processing is done after the end of the time window, there is high latency for receiving the processed results. For example, if the time window is a day, then the data will not be processed until after the day is done. Further, if any data from one time period is delayed in being provided to the partition file system until after the end of the time period, that data will typically not be processed until the end of the next time window adding additional latency. An example of this is in the banking industry where a deposit received on a given day after <NUM> PM is added into the deposits for the next day. Thus, a batch processing system provides large data volume processing, but at a cost of delayed processing results.

Since both of the stream processing system <NUM> and the batch processing system <NUM> provide different advantages, it is common for a company to implement both systems. Thus, these companies are required to have separate, dedicated resources for both systems. In addition, it quite common for the data that is to be processed to come from the same data producers and thus have to be duplicated so that both the stream processing system and the batch processing system are able to process the data in their respective ways. The need for two separate processing systems with their own dedicated resources can be costly.

Advantageously, the embodiments disclosed herein provide for the creation of a near-real-time stream processing system that uses the same input files as a batch processing system. By sharing the same input files, the need for additional storage and resources required by the messaging-queue based stream processing system described previously is eliminated. Accordingly, resource costs are reduced significantly since two dedicated systems are no longer needed. This is illustrated in <FIG>, which illustrates an embodiment <NUM> of a processing system. As illustrated in the figure, data producers <NUM>, <NUM> and potentially any number of producers <NUM>, which may correspond to the previously described data producers provide data to a partition file system <NUM>. The partition file system <NUM>, which may correspond to the partition file system <NUM>, receives the data and generates the data containers and partition files (not illustrated). The data in the partition files may then be processed by the batch processing system <NUM> and a near-real-time stream processing system <NUM>. The partition file system <NUM> and the near-real-time stream processing system <NUM> will be described in more detail to follow. The operation of the batch processing system <NUM> will now be described. For example, in the batch processing system <NUM>, the data container <NUM> and its partition files <NUM> and <NUM> are accessed at the end of a defined partition window. The defined partition window will be described in more detail to follow. One or more batches of the data included in the partition files <NUM> and <NUM> are generated. Processing of the one or more batches of data may then be performed.

In this way, the processing system is configured to implement both the near-real-time stream processing system <NUM>, as will be described in more detail to follow, and the batch processing system <NUM>. Thus, in the processing system <NUM> the same data containers and the one or more partition files that are included in the data containers (i.e., data container <NUM> and partition files <NUM>-<NUM> and data container <NUM> and partition files <NUM>-<NUM>) are accessible by the near-real-time stream processing system <NUM> and by the batch processing system <NUM>. In some embodiments, the same data containers and the one or more partition files that are included in the data containers are accessible by both the near-real-time stream processing system <NUM> and by the batch processing system <NUM> at a same time or, alternatively, at a different time period. That is, the batch processing system <NUM> and the near-real-time stream processing system <NUM> may simultaneously process the data in the partition files of the various data containers. Alternatively, the batch processing system <NUM> and the near-real-time stream processing system <NUM> may process the data in the partition files of the various data containers at different times. <FIG> illustrates an embodiment of a near-real-time stream processing system <NUM> such as the near-real-time stream processing system <NUM>. As illustrated, the near-real-time stream processing system <NUM> includes a configuration data store <NUM>. The configuration data store <NUM> may be any reasonable storage as circumstances warrant. In operation, the configuration data store <NUM> stores various configurations that are specify how the near-real-time stream processing system <NUM> is to operate. These configurations are typically configurable by a user of the system according to their operational needs. In addition, as will be described in more detail to follow, the task manager <NUM> uses the configuration data store to record the progress status of various processing tasks performed by the system.

As illustrated, the configuration data store <NUM> stores a partition window configuration <NUM>. The partition window configuration is a time range or period that controls when data is to be included in a data container and its respective partition files. For example, the time range may be a day, an hour, or any other reasonable time range. As will be described in more detail to follow, data received by the system during the partition window is appended to one or more partition files in the data container corresponding to the partition window. In some embodiments, the partition window is defined by a window start time t1 and window size. For example, the window start time may be defined as the current date at <NUM>:<NUM>:<NUM> and the window size may be defined as <NUM> hours. Accordingly, if the current date were January <NUM>, <NUM>, data received on that date for <NUM> hours will be included in partition files included in a data container for that day. When the clock changes to <NUM>:<NUM>:<NUM> on January <NUM>, <NUM>, then data received on that date for <NUM> hours will be included in partition files included in a different data container for January <NUM>, <NUM>. This process would repeat for each new partition window of <NUM> hours. The configuration data store <NUM> also stores a partition lifetime window configuration <NUM>. The partition lifetime window configuration <NUM> is a time range or period that controls how long partition files are considered active so that processing can be performed on the partition files and thus functions as a measure of processing time. The time range may be the time range of the partition window plus an additional amount of time. Thus, if the partition window is one day, then perhaps the partition lifetime window may be <NUM> hours, which is six hours longer than the partition window. For example, the partition lifetime window may be defined by the partition window start time (i.e., January <NUM>, <NUM> in the above example) at <NUM>:<NUM>:<NUM>, the partition window size of <NUM> hours, and the additional amount of time of six hours (i.e., January <NUM>, <NUM> at <NUM>:<NUM>:<NUM> end time) to reach the <NUM>-hour size. When the clock changes to January <NUM>, <NUM> at <NUM>:<NUM>:<NUM>, then the partition lifetime window for the data container and its partition files generated on that date would be until January <NUM>, <NUM> at <NUM>:<NUM>:<NUM>. This would repeat for each new partition window. Of course, other amounts of time different from six hours may be used as the additional amount of time as circumstances warrant. Accordingly, since the partition lifetime window has a time range that equals the time range of the partition window plus the additional amount of time, the partition lifetime window is larger than the partition window.

In addition, the configuration data store <NUM> stores a task size configuration <NUM>. In many embodiments the size of a partition file may be a few Gigabytes to several hundred Gigabytes. Accordingly, the task size configuration <NUM> is used to define a subset of the data that is to be processed during a given processing iteration. For example, in one embodiment the task size may be between 20MB and 200MB of data, although other task sizes may be chosen. As will be appreciated, the size of the task size configuration <NUM> helps determine the overall latency of the system. For example, the larger the size of the data subset, then the longer the latency will be as the system will wait for the subset of data corresponding to the task size configuration to be appended to the partition files before processing it. Thus, the size of the data subset may be set for a <NUM> second latency, a <NUM> second latency, a minute latency, or some other desired latency. Accordingly, it may be advantageous to avoid configuring the task size to be too large to avoid an undesirable long latency. On the other hand, if the subset of data in the task size configuration <NUM> is set too small, then efficiency may be lowered as the system may process a small amount of data. It will be appreciated that since there is some latency, usually at least a <NUM> second latency, for the subset of the data to be processed during a given processing iteration (i.e., processing task), the embodiments disclosed herein may be considered near-real-time stream processing system. In some embodiments, the task size configuration <NUM> may also define a timeout. The timeout specifies how long a task should be considered active. If the task size is not met within the defined timeout, a new, smaller task will be generated. The configuration data store <NUM> further stores a processing model configuration <NUM>. As will be explained in more detail to follow, the processing model configuration <NUM> allows for the selection of the type of processing to be performed on the data in the partition files. The ellipses <NUM> illustrate that there may be any number of addition configurations stored in the configuration data store <NUM> as circumstances warrant.

As illustrated n <FIG>, the near-real-time stream processing system <NUM> receives data such as telemetry data from a data producer <NUM>, a data producer <NUM>, and any number of additional data producers as illustrated by the ellipses <NUM>. In operation, the various data producers, which may be service endpoints or client endpoints, generate data or events that are to be processed by the near-real-time stream processing system <NUM>.

The received data is then provided to a partition file system <NUM>, which may correspond to the partition file system <NUM>. Although not illustrated, the partition file system <NUM> may include an ingestion pipeline that delivers the data from the producers <NUM>-<NUM> and that generates the various partition files. As illustrated, the partition file system <NUM> generates a logical data container <NUM> based on the time range defined in the partition window configuration <NUM>. As discussed previously, the partition window may be a day, an hour, or some other reasonable time range.

The partition file system <NUM> generates partition files in the data container <NUM>. For example, at a beginning of the partition window, a partition file <NUM> and a partition file <NUM> may be generated. The partition file <NUM> includes metadata <NUM> about the partition file and the data <NUM> that is appended to the partition file. The metadata <NUM> includes a time stamp 423a that indicates a time and/or date that can be used in conjunction with the partition window <NUM> to determine what data container the partition file should be placed. For example, if the data container is for the entire day of January <NUM>, <NUM>, then the time stamp 423a would indicate the date as January <NUM>, <NUM>. A file identification (ID) 423b identifies the file and may be a file Uniform Resource Locator (URL) that identifies the location of the partition file in the distributed file system. File size metadata 423c specifies the size of the partition file and is updated every time new data is appended to the partition file, for example when data 424a is appended to the data <NUM> at a later time. Last update metadata 423d specifies the time that the partition file was last updated and is updated every time new data is appended to the partition file such as when data 424a is appended to the data <NUM>. The ellipses 423e illustrate that the partition file <NUM> may have additional metadata.

The partition file <NUM> includes metadata <NUM> about the partition file and the data <NUM> that is appended to the partition file. The metadata <NUM> includes a time stamp 426a that indicates a time and/or date that can be used in conjunction with the partition window <NUM> to determine what data container the partition file should be placed. For example, if the data container is for the entire day of January <NUM>, <NUM>, then the time stamp 426a would indicate the date as January <NUM>, <NUM>. A file identification (ID) 426b identifies the file and may be a file Uniform Resource Locator (URL) that identifies the location of the partition file in the distributed file system. File size metadata 426c specifies the size of the partition file and is updated every time new data is appended to the partition file. Last update metadata 426d specifies the time that the partition file was last updated and is updated every time new data is appended to the partition file. The ellipses 426e illustrate that the partition file <NUM> may have additional metadata.

The near-real-time stream processing system <NUM> includes a task manager <NUM>. In operation, the task manager <NUM> is responsible for partition file discovery, partition file management, and task management. As shown in <FIG>, the task manager <NUM> scans or reads the metadata of each of the partition files. For example, the task manager <NUM> scans or reads the metadata <NUM> of the partition file <NUM> and the metadata <NUM> of the partition file <NUM>. The task manager <NUM> uses the metadata <NUM> and <NUM>, in particular the time stamps 423a and 426a and the last update metadata 423d and 426d, to determine if the partition files are within the partition lifetime window <NUM>. In other words, the task manager <NUM> determines that the partition lifetime window <NUM> has not yet expired. If the partition lifetime window has not yet expired, then the partition files <NUM> and <NUM> are considered active partition files and the task manager <NUM> assigns processing tasks for the partition files as will be explained in more detail to follow. The near-real-time stream processing system <NUM> includes task worker instances <NUM>. The task worker instances <NUM> represent multiple processing instances that can be distributed across a number of virtual machines and other processors. For example, the task worker instances <NUM> may include a task worker instance <NUM>, a task worker instance <NUM>, and any number of additional task worker instances as illustrated by ellipses <NUM>. As will be explained in more detail to follow, each of the task worker instances <NUM> are assigned, for a given time, zero, one, or multiple tasks by the task manager <NUM>. The task worker instances are then able to access the data <NUM> and <NUM> in the partition files <NUM> and <NUM> and process the data.

<FIG> illustrates an embodiment of sequential processing of the partition file <NUM>. As illustrated, in the embodiment the task manager <NUM> reads the metadata <NUM> to determine if the partition file <NUM> is active (i.e., within the partition lifetime window <NUM>). In the embodiment, the partition file <NUM> is active and so the task manager <NUM> records in the configuration data store <NUM> that the partition file <NUM> is active as shown at <NUM> and also records the file size metadata 423c. The task manager <NUM> accesses the processing model configuration <NUM> to determine that sequential processing is to be performed. In other embodiments, however, the task manager <NUM> is able to dynamically choose the processing model to use without the need for the processing model configuration <NUM> based on the size of the various partition files. The task manager uses the file size metadata 423c and a checkpoint record (i.e., checkpoint record <NUM>) to create a processing task <NUM>. The processing task <NUM> includes the file ID 423b. The processing task <NUM> also includes a continuous range of data in the partition file <NUM> that is defined by a start position <NUM> and an end position <NUM>. As discussed previously, the size of the continuous range (and all continuous ranges discussed herein) is defined by the task size configuration <NUM>.

The task manager <NUM> assigns the task <NUM> to the task worker instance <NUM> of the task worker instances <NUM>. The task worker instance <NUM> directly accesses the continuous data range defined by the start position <NUM> and the end position <NUM> and performs the appropriate processing on this data. It will be noted that the task manger <NUM> does not access the actual data in the partition file <NUM> (or any of the partition files), but only reads the metadata as discussed. This allows for the task manager <NUM> to be implemented with low resource cost. In addition, since only the task worker instances access the actual data, different types of data specific logic can be used by the task worker instances while still maintaining the implementation and design of the task manager <NUM>.

Once the task worker instance <NUM> has completed the processing task <NUM>, it reports this to the task manager <NUM>. The task manager <NUM> will then track the progress of the processing for partition file <NUM> in the configuration data store <NUM>. The progress record may be a checkpoint record <NUM> that indicates the address range of the data that has been processed. In the current embodiment, the checkpoint record <NUM> would indicate that the address range of the data between the start position <NUM> and the end position <NUM> has been processed as denoted by 457a. Since sequential processing of the partition file <NUM> is being performed, there is only one active task at any given time. Accordingly, once the processing task <NUM> is completed, the task manager <NUM> creates a second processing task <NUM>. The second processing task <NUM> includes the file ID 423b for the partition file <NUM>. The second processing task <NUM> also includes a continuous range of data in the partition file <NUM> that is defined by a start position <NUM> and an end position <NUM>. In addition, the task manager <NUM> uses the checkpoint record 457a to ensure that the data defined by the start position <NUM> and an end position <NUM> has not already been processed.

Although <FIG> shows that the second processing task <NUM> being assigned to the task worker instance <NUM> of the task worker instances <NUM>, this is for ease of illustration only. The dashed lines around the second processing task <NUM> illustrate that the second processing task <NUM> may be assigned to a different task worker instance <NUM> than the worker instance <NUM>. The task worker instance <NUM> (or the other task worker instance) directly accesses the continuous data range defined by the start position <NUM> and the end position <NUM> and performs the appropriate processing on this data.

Once the task worker instance <NUM> (or the other task worker instance) has completed the processing task <NUM>, it reports this to the task manager <NUM>. The task manager <NUM> than updates the checkpoint record <NUM>. In the current embodiment, the checkpoint record would be updated to indicate that the address range of the data between the start position <NUM> and the end position <NUM> has been processed as denoted by 457b. This is because in sequential processing, the end position of a previous processing task (i.e., end position <NUM> of processing task <NUM>) is typically the same as the starting position of a subsequent processing task (i.e., start position <NUM> of second processing task <NUM>). Thus, the task manager <NUM> merges the checkpoint 457a with the updated progress record into the checkpoint 457b.

Once the second processing task <NUM> is completed, the task manager <NUM> may continue to scan the partition file <NUM> for file size and update changes and continue to create additional processing tasks as needed. This is illustrated by the ellipses and processing task N <NUM>. The processing tasks up to processing task N <NUM> are then provided to a task worker instance for processing in the manner previously discussed and the checkpoint record <NUM> is updated accordingly as denoted by the ellipses 457c. It will be appreciated that illustration of the checkpoint records 457a, 457b, and 457c is for ease in showing that the task manager <NUM> continually updates the checkpoint record <NUM>. However, it will be appreciated that there is typically only one checkpoint record that is kept in the configuration store <NUM> at a time. Thus, when the checkpoint record 457a is updated to checkpoint record 457b, the checkpoint record 457a is removed from the configuration store and when the checkpoint record 457b is updated to the checkpoint record 457c, then the checkpoint record 457b is removed and so on every time the checkpoint record <NUM> is updated.

As discussed previously, the address range for processing the data in the partition file <NUM> is defined by the start position and the end position of each processing task. Thus, it may be possible that the start position or the end position for any given processing task is in the middle of one data record. Accordingly, in some embodiments the data records may be formatted to be splitable. In such embodiments, the task worker instances <NUM> are able to read forwards and backwards in the data to find the indicated start position or end position. For example, if the end position is in the middle of one record, the task worker instances <NUM> will either discard the last partial record or read a little more to get a complete record. In either case, the checkpoint will be adjusted to the end of the record in the configuration data store <NUM>. Given the checkpoint can be adjusted to the end position of one record, there is typically not the problem that start position is in the middle of one record. However, in case there is corruption, the task worker instances <NUM> will move backward to find the current record's start position. In other embodiments, it is possible that the task size specified by the task size configuration <NUM> is larger than the remaining data in the partition file <NUM>. Accordingly, in such embodiments the task manager <NUM> may create a processing task that only includes the remaining unprocessed data in the partition file. In other embodiment, the task manager <NUM> may wait until more data such as data 424a is appended to the partition file <NUM> so that there is enough data to create a processing task that meets the task size configuration <NUM>. In still other embodiments, the task manager <NUM> may wait a predetermined amount of time and then create the processing task that only includes the remaining unprocessed data in the partition file. Although not illustrated, sequential processing may also be performed on the partition file <NUM>. Thus, the task manager <NUM> records in the configuration data store <NUM> that the partition file <NUM> is active as shown at <NUM> and also records the file size metadata 426c. A checkpoint record <NUM> is also recorded and updated as needed to track the processing of the partition file <NUM>. <FIG> illustrates an embodiment of parallel processing of the partition file <NUM>. As with the embodiment of <FIG>, the task manager <NUM> determines that the partition file is still active and then creates the processing tasks <NUM>, <NUM><NUM>, and up to processing task N <NUM> to include the file ID and the start and end positions as previously described in relation to <FIG>. The task manager <NUM> may select parallel processing based on the processing model configuration <NUM> or dynamically based on the processing load.

In parallel processing, the task manager <NUM> simultaneously creates the processing tasks <NUM>, <NUM>, and up to processing task N <NUM>. Each of the processing tasks is assigned to multiple task worker instances. For example, the processing task <NUM> is assigned to the task worker instance <NUM>, the processing task <NUM> is assigned to the task worker instance <NUM>, and the processing task N <NUM> is assigned to the task worker instance N. In some embodiments, a threshold may set an upper limit to the number of parallel processing tasks that can be created at once so as to ensure efficient use of the processing resources of the near-real-time stream processing system <NUM>. In addition, this helps ensure that checkpoint records per partition can be managed without causing performance issues in the configuration data store <NUM>.

The task worker instances <NUM>, <NUM>, and N access the data in the partition file according to data range defined by their respective start and end positions and perform the appropriate processing on the data. The task worker instances <NUM>, <NUM>, and N report to the task manager <NUM> when they have completed their respective processing tasks.

The task manager <NUM> records the progress in the configuration data store <NUM>. Since multiple task worker instances are reporting at different times, the task manager <NUM> may record more than one checkpoint in the configuration data store <NUM>. For example, a checkpoint <NUM> and a checkpoint <NUM> may be recorded for the processing done for the processing tasks <NUM> and <NUM> respectively since it is possible that the data ranges of both of these processing tasks will not be continuous. The task manager <NUM> will check to see if an end position of one checkpoint is equal to the start position of another checkpoint. If this is found, then the task manager will merge the two checkpoints as shown at <NUM>. In this way, it is possible to determine the overall processing progress of the partition file. Although not illustrated, parallel processing may also be performed on the partition file <NUM>.

In some instances, a "hot" partition or data skew issue may occur when a much larger amount of data is placed in one partition file then is placed in the other partition files. In such instances, attempts to process the hot partition may slow the system down. Advantageously, parallel processing as described herein can provide at least a partial solution to this problem. Configuring parallel processing for the hot partition can quickly process the backlog in the hot partition without slowing the system down too much.

<FIG> illustrates an embodiment of parallel processing by the same working instance of the partition file <NUM>. As with the embodiment of <FIG>, the task manager <NUM> determines that the partition file is still active and then creates the processing tasks <NUM>, <NUM>, and up to processing task N <NUM> to include the file ID and the start and end positions as previously described in relation to <FIG>. The task manager <NUM> may select parallel processing by the same working instance based on the processing model configuration <NUM> or dynamically based on the processing load.

In parallel processing by the same working instance, the task manager <NUM> simultaneously creates the processing tasks <NUM>, <NUM>, and up to processing task N <NUM>. Each of the processing tasks is assigned to the task worker instance <NUM>. The task worker instance <NUM> then decides how to parallel process the processing tasks.

Once the processing tasks have been completed, the task worker instance <NUM> reports to the task manager <NUM>. The task manager <NUM> then records the checkpoints in the configuration data store <NUM> as described in relation to <FIG>. Although not illustrated, parallel processing by the same working instance may also be performed on the partition file <NUM>.

<FIG> further illustrates how the task manager <NUM> is able to dynamically at runtime discover new partition files. For example, suppose that partition files <NUM> and <NUM> are created by the partition file system <NUM> at the time that the data container <NUM> is created. The task manager <NUM> will scan the data container <NUM> and discover the partition files <NUM> and <NUM>. The task manager <NUM> will then mark that these partition files are active in the configuration data store <NUM> as discussed previously since they are within the current partition lifetime window <NUM>.

The task manager <NUM> will continue to periodically scan the data container <NUM>. While scanning the data container <NUM> at a time subsequent to the time the data container is created, the task manager <NUM> may discover that the partition file <NUM> has been added to the data container <NUM>. This is illustrated by the dashed lines around partition file <NUM>. The task manager <NUM> will then mark that this partition file is active in the configuration data store <NUM> since it is within the current partition lifetime window <NUM>. The task manager <NUM> will further read the metadata 428a of the partition file <NUM> to create and assign processing tasks for the data 425b of the partition file <NUM> and will record and update checkpoints for this partition file as previously discussed. The ellipses <NUM> illustrate that the task manager <NUM> may discover any number of additional partition files as they are created by the partition file system <NUM> during a given partition lifetime window. Accordingly, the embodiments disclosed herein do not require that the number of partitions be known ahead of runtime. Rather, the task manager <NUM> will discover any newly added partition files anytime it scans the data container <NUM> during the partition lifetime window <NUM>.

In some embodiments, there may be a system failure or other processing delay that interrupts the near-real-time stream processing system <NUM>. In such cases, when the task manager <NUM> is restarted, it loads all partitions that are marked as being active in the configuration data store <NUM>. Since the last updated checkpoint will also be stored in the configuration data store <NUM>, the task manager <NUM> will know where to pick up when assigning the processing tasks. This advantageously ensures that all data is process as needed. For example, suppose that there was an issue causing a processing delay for the partition file <NUM> and the current time stamp was for January <NUM>, <NUM> at <NUM>:<NUM>:<NUM> when the processing delay occurred. The task manager can still discover the partition file <NUM> from the active record in the configuration data store <NUM> and will keep it active until all the data is properly processed even if the current time stamp is January <NUM>, <NUM> at <NUM>:<NUM>:<NUM> when the task manager <NUM> is restarted.

<FIG> illustrates a further embodiment of the near-real-time stream processing system <NUM>. As illustrated, the partition file system <NUM> has discovered the partition files <NUM>, <NUM>, <NUM>, and potentially <NUM> during the partition window <NUM> for the data container <NUM>. In the embodiment of <FIG>, however, the time has changed so that a new partition window <NUM> is now active. In other words, suppose the partition window <NUM> for the data container <NUM> and its partition files was January <NUM>, <NUM> for <NUM> hours. Then the new partition window <NUM> would be January <NUM>, <NUM> for <NUM> hours. Accordingly, at time <NUM>:<NUM>:<NUM> on January <NUM>, <NUM> the partition file system <NUM> will create the data container <NUM> and also create a partition file <NUM>, partition file <NUM>, and potentially any number of additional partition files as illustrated by ellipses <NUM>. Data received from the data producers <NUM>-<NUM> received after time <NUM>:<NUM>:<NUM> on January <NUM>, <NUM> will be placed in the partition files <NUM>, <NUM>, and potentially <NUM>. The partition file <NUM> will include metadata <NUM> and data <NUM>. The partition file <NUM> will include metadata <NUM> and data <NUM>. The metadata <NUM> and <NUM> may include the same types of metadata as discussed previously in relation to partition files <NUM> and <NUM>.

The task manager <NUM> will begin to scan the data container <NUM> and will dynamically, at runtime (i.e., the task manger <NUM> does not know ahead of time the number of partition fields in the data container <NUM>) discover the partition files <NUM>, <NUM>, and potentially the partition files <NUM>. Suppose, however, that the partition lifetime window <NUM> for the data container <NUM> and its partition files was <NUM> hours, then this partition lifetime window would be active until <NUM>:<NUM>:<NUM> on January <NUM>, <NUM>. Accordingly, from <NUM>:<NUM>:<NUM> until <NUM>:<NUM>:<NUM> on January <NUM>, 2020the task manager <NUM> will consider that the data container <NUM> and its partition files are still active and will continue to scan these partition files for any changes in the file size and last update metadata. Although data received after time <NUM>:<NUM>:<NUM> on January <NUM>, <NUM> will be placed in the data container <NUM>, there may be processing delays or the like that cause that some data that was received before the end of the <NUM>-hour partition window of January <NUM>, <NUM> to not be placed into one of the partition files of the data container <NUM> until after the end of that partition window or alternatively to not be processed before the end of the partition window. This is illustrated by data 424b that is appended to the partition file <NUM>. Accordingly, having the partition lifetime window be larger than the partition window allows for the late arriving data to be processed with the January <NUM>, <NUM> data and not the January <NUM>, <NUM> data.

<FIG> shows that the task manager <NUM> periodically scans the partition file <NUM> and reads the metadata <NUM> since this partition file is still active. For ease of illustration, the ellipses 430A represent that the task manager <NUM> also periodically scans the partition file <NUM>, <NUM>, and potentially <NUM> and reads their respective metadata since these partition files are also considered active since the partition lifetime window <NUM> for the data container <NUM> and its partition files has not ended.

<FIG> also shows that the task manager <NUM> periodically scans the partition file <NUM> and reads the metadata <NUM> since this partition file is active since the partition window <NUM> for the data container <NUM> and its partition file is the current partition window. For ease of illustration, the ellipses 430B represent that the task manager <NUM> also periodically scans the partition file <NUM> and potentially <NUM> and reads their respective metadata since these partition files are included in the current partition window <NUM>.

<FIG> illustrates sequential processing while the partition lifetime window <NUM> for the data container <NUM> and its partition files are still active. As shown in <FIG>, the task manager <NUM> creates a processing task <NUM> for the partition file <NUM> that includes an address range (not illustrated) that includes the data 424b. The task manager <NUM> may assign the processing task <NUM> to the task worker instance <NUM>. The task worker instance <NUM> may then access the data 424b and perform the appropriate processing and report the progress back to the task manager <NUM> in the manner previously discussed. The task manager <NUM> may then record an updated checkpoint record <NUM> to reflect the processing of the data 424b. It will be noted that the configuration data store <NUM> will continue to show that the partition file <NUM> is active since the partition lifetime window <NUM> for the data container <NUM> and its partition files has not ended. As shown in <FIG>, the task manager <NUM> also creates a processing task <NUM> for the partition file <NUM> that includes an address range (not illustrated) that includes a subset of the data <NUM> according to the task size configuration <NUM>. The task manager <NUM> may assign the processing task <NUM> to the task worker instance <NUM>. The task worker instance <NUM> may then access the data <NUM> included in the address range and perform the appropriate processing and report the progress back to the task manager <NUM> in the manner previously discussed. The task manager <NUM> may then record a checkpoint <NUM> to reflect the processing of the data <NUM>. The task manager <NUM> will also record that the partition file <NUM> is active in the configuration data store <NUM> as denoted at <NUM>. The ellipses <NUM> represent that the task manager <NUM> may create and assign any number of the additional processing tasks as needed for any of the other active partition files. The ellipses <NUM> and <NUM> are for ease of illustration and represent that the task manager <NUM> may record in the configuration data store <NUM> checkpoint information and active status information about any of the other active partition files of the data container <NUM> and data container <NUM>, respectively.

Suppose that the timestamp shows that the time is <NUM>:<NUM> :<NUM> on January <NUM>, <NUM>. The task manager <NUM> determines that the partition lifetime window <NUM> for the data container <NUM> and its partition files has ended based on the time stamp. The task manager will scan the file size metadata 423c for the partition file <NUM>, the file size metadata 426c for the partition file <NUM>, and the file size metadata for the partition file <NUM> and any of the additional partition files <NUM>. Using the file size metadata, the task manager determines if all of the data corresponding to the file size has been appropriately processed. If the data has been appropriately processed the data container <NUM> and its partition files will be closed.

In addition, the information about the partition files of the data container <NUM> will be removed from the configuration data store <NUM>. This is shown in the view of the configuration data store <NUM> shown in <FIG> where only information about the partition files of the data container <NUM> are shown as being recorded in the configuration data store <NUM>. Thus, when a partition lifetime window ends, any record of the partition files associated with the partition lifetime window that indicate that the partition window is active are removed from the configuration store <NUM>.

<FIG> illustrates a flow chart of an example method <NUM> for implementing a near-real-time stream processing system using the same distributed file system as a batch processing system. The method <NUM> will be described with respect to one or more of <FIG> discussed previously.

The method <NUM> includes generating <NUM> a data container and one or more partition files within the data container according to a defined partition window. The defined partition window specifies a first time range that controls when data is to be included in the one or more partition files of the data container. For example, as previously discussed the partition file system <NUM> generates the data container <NUM> and the partition files <NUM>, <NUM>, <NUM>, and potentially <NUM> and generates the data container <NUM> and the partition files <NUM>, <NUM>, and potentially <NUM>. The partition files <NUM>, <NUM>, <NUM>, and potentially <NUM> are generated during the partition window <NUM> for the data container <NUM> and its partition files. The partition files <NUM>, <NUM>, and potentially <NUM> are generated during the partition window <NUM> for the data container <NUM> and its partition files. The partition window <NUM> may be time range such as a day, an hour, or some other time period and data is appended to the partition files during this time range.

The method <NUM> includes scanning <NUM> the data container to determine if the one or more partition files are within a defined partition lifetime window. The defined partition lifetime window specifies a second time range that controls how long the one or more partition files are considered active such that processing is to be performed on the one or more partitions. For example, as previously discussed the task manager <NUM> scans the partition files <NUM>, <NUM>, <NUM>, and potentially <NUM> and the partition files <NUM>, <NUM>, and potentially <NUM> to determine if the partition files are within the partition lifetime window <NUM>. The partition files that are within the partition lifetime window <NUM> are considered active for processing. The partition lifetime window <NUM> is the time length of the partition window plus some additional amount of time.

The method <NUM> includes, for each partition file within the defined lifetime window, creating <NUM> one or more processing tasks based on an amount of data included in the one or more partition files. For example, as previously discussed the task manager <NUM> creates the processing tasks <NUM>, <NUM>, <NUM>, and <NUM>. The processing tasks may be based on the task size configuration <NUM>.

The method <NUM> includes accessing <NUM> the data in the one or more partition files and performing the one or more processing tasks. For example, as previously discussed the task worker instances <NUM> directly access the data in the partition files and process the data according to the processing tasks.

The method <NUM> includes recording <NUM> in a configuration data store information about the one or more partition files. For example, as previously discussed the checkpoint records <NUM>, <NUM>, <NUM>-<NUM>, and <NUM> may be stored in the configuration data store <NUM>. In addition, other information such as the active status <NUM>, <NUM>, and <NUM> may also be recorded in the configuration data store <NUM>.

For the processes and methods disclosed herein, the operations performed in the processes and methods may be implemented in differing order. Furthermore, the outlined operations are only provided as examples, and some of the operations may be optional, combined into fewer steps and operations, supplemented with further operations, or expanded into additional operations without detracting from the essence of the disclosed embodiments.

Claim 1:
A computing system (<NUM>) implementing a distributed file system (<NUM>, <NUM>) that is configured for use by a batch processing system (<NUM>), the computing system implementing a near-real-time stream processing system (<NUM>) using the same distributed file system (<NUM>, <NUM>) as the batch processing system, the computing system comprising:
one or more processors (<NUM>); and
one or more computer readable hardware storage devices (<NUM>) having computer executable instructions (<NUM>) stored thereon that when executed by the one or more processors cause the computing system to:
generate (<NUM>) a data container (<NUM>, <NUM>) and one or more partition files (<NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>) within the data container according to a defined partition window (<NUM>), the defined partition window specifying a first time range that controls when received data (<NUM>, <NUM>, 428b, <NUM>, <NUM>) is to be included in the one or more partition files of the data container, wherein the one or more partition files are generated during the partition window, the data received during the partition window being placed into the one or more partition files;
after the end of the partition window, performing batch processing on the data in the one or more partition files by the batch processing system;
by the near real-time steam processing system, performing near real-time stream processing by:
- periodically scanning (<NUM>) the data container to determine if the one or more partition files are within a defined partition lifetime window (<NUM>), the defined partition lifetime window specifying a second time range that controls how long the one or more partition files are considered active such that processing is to be performed on the one or more partitions,
- for each partition file within the defined lifetime window, creating (<NUM>) one or more processing tasks (<NUM>, <NUM>, <NUM>, <NUM>) based on an amount of data (<NUM>) included in the one or more partition files, and
- accessing (<NUM>) the data in the one or more partition files and performing the one or more processing tasks thereon, the accessing of the data by the near-real-time processing system being from the same one or more partition files in the same container as the data accessed by the batch processing system to perform the batch processing; and
record in a configuration data store information about the one or more partition files.