Anchor shortening across streaming nodes

A method for facilitating anchor shortening across streaming nodes in an event stream processing system may include receiving a full anchor at an upstream marshaller. The full anchor may be associated with a data batch that corresponds to one or more event streams. The full anchor may include an indication of an input point for the one or more event streams. The full anchor may be received from an upstream compute processor. The method may also include mapping the full anchor to an index anchor and passing the index anchor to a downstream marshaller.

BACKGROUND

Event stream processing refers to the process of quickly analyzing time-based data, such as time series data. In the context of event stream processing, the term “event” refers to any occurrence that happens at a defined time and can be recorded using one or more data fields. An “event stream” is a sequence of multiple data events, which may be ordered by time. Some examples of different types of event streams include Internet of Things (IoT) data (e.g., sensor data, signals from control systems, location information), business transactions (e.g., customer orders, bank deposits, invoices), and information reports (e.g., social media updates, market data, weather reports).

With traditional approaches, data is typically processed after it has been stored. Advantageously, event stream processing allows data to be analyzed as it is being created and before it is stored. For example, data may be analyzed when it is streaming from one device to another. This allows for faster reaction time and may even provide an opportunity for proactive measures to be taken.

Event stream processing may be utilized to identify meaningful patterns or relationships within event streams in order to detect relationships like event correlation, causality, or timing. There are a wide variety of industries that can benefit from event stream processing, including network monitoring, cybersecurity, retail optimization, surveillance, fraud detection, financial trading, and e-commerce.

DETAILED DESCRIPTION

From time to time, one or more nodes in an ESP system may fail. Some ESP systems may simply skip any data that was missed due to node failure. However, it may be desirable for the node(s) in an ESP system to be able to resume processing from the point at which failure occurred.

One technique to recover from node failure without loss or duplication of results relies on the formation of anchors. In this context, the term “node” may refer to a single computing device or to a set of computing devices that collectively function to perform one or more tasks. An ESP system may include multiple nodes. The term “anchor” refers to a logical pointer into an event stream that allows processing within an ESP system to resume from a precise point. A general methodology for constructing systems of anchors is described in U.S. patent application Ser. No. 14/732,416, titled “Using Anchors for Reliable Stream Processing,” filed on Jun. 5, 2015, and assigned to the owner of the present application.

An anchor may take the form of metadata that is associated with a data batch. An anchor corresponding to a particular data batch may include an indication of a latest input point for every input that is used to generate the data batch. An anchor may also include some additional wrapping. In this context, the term “data batch” refers to a set of data that corresponds to one or more event streams. The term “input” refers to a source of an event stream (e.g., a sensor, a control system, a software application). The term “input point” refers to a particular point in one or more event streams corresponding to a data batch. An input point may be characterized in a variety of different ways. Some examples of input points include a file offset, a file identifier, and a time stamp.

As used herein, the terms “upstream” and “downstream” may be used to describe the flow of data in an ESP system. An ESP system may be configured so that a downstream entity pulls data from an upstream entity via a pull operation. As part of performing the pull operation, the downstream entity may reference a specific anchor corresponding to a particular data batch. The anchor indicates a point in the data that the response should come after.

More specifically, a downstream entity may send a request for a batch of data to an upstream entity. The request may reference a particular anchor. In response to the request, the upstream entity may return the requested batch of data, beginning at the point in the data corresponding to the specified anchor. The upstream entity may also return a new anchor. When the downstream entity makes a subsequent request for another batch of data from the upstream entity, the request may reference the new anchor.

Anchors may be nested, such that a particular anchor may include a reference to one or more other anchors. Consider a relatively simple example that involves three processing entities, which will be referred to as an input processor, a first compute processor, and a second compute processor. In this context, the term “input processor” refers to a processing entity that ingests one or more event streams from one or more inputs. The term “compute processor” refers to a processing entity that performs one or more processing operations on event stream(s).

Suppose that these processing entities are connected serially. Further suppose that the input processor ingests data from a data source, the first compute processor pulls data from the input processor, and the input processor returns anchor a1along with the requested data. Subsequently, when the second compute processor pulls data from the first compute processor, suppose that the first compute processor returns anchor a2along with the requested data. In this example, anchor a2may include a reference to anchor a1. Thus, anchor a2may be considered to be a nested anchor. More generally, as used herein, the term “nested anchor” refers to an anchor that includes a reference to at least one other anchor.

The previous example was relatively simple and involved just three processing entities. However, the overall structure (or topology) of an ESP system may be fairly complex. For example, an ESP system may include multiple nodes, and these nodes may be interconnected in various ways. Moreover, a single node may include multiple processing entities, which may also be interconnected in various ways.

The size of an anchor may grow exponentially with the number of processing entities that it covers. Thus, in an ESP system that has a complex processing topology, the size of anchors can grow quickly. In the simple example discussed above, anchor a2was a function of just one other anchor, namely anchor a1. But in an ESP system with a complex processing topology, an anchor could be a function of a very large number (e.g., thousands) of other anchors.

As anchors become larger and more complex, the value of using anchors may be diminished. At some point, the size of an anchor may become so large that problems may occur. For example, since ESP systems are memory intensive, the use of large anchors might exhaust the available supply of memory.

The present disclosure is generally related to facilitating anchor shortening across streaming nodes in an ESP system to support complex processing topologies. In accordance with the present disclosure, boundary entities may be provided within the nodes of an ESP system. These boundary entities may be referred to herein as “marshallers.” Two different types of marshallers are described herein: downstream marshallers and upstream marshallers. In general terms, a marshaller is an entity that is configured to facilitate anchor mapping, as will be described in greater detail below. An upstream marshaller within a particular node may be configured to perform anchor mapping and remapping. A downstream marshaller within a particular node may pull data from an upstream marshaller in another node.

More specifically, when an upstream marshaller receives an anchor, the upstream marshaller may map the anchor to an index (e.g., a single numeric value). Instead of passing the anchor (which may be quite complex) to a downstream marshaller, the upstream marshaller may pass the index instead. This limits the complexity of anchors within an ESP system. The upstream marshaller may store mapping information, which may be any information that indicates the relationship between the anchor and the index. If the downstream node subsequently fails and the downstream marshaller requests a batch of data corresponding to the index, the upstream marshaller may use the mapping information to determine the original anchor.

FIG. 1illustrates an example of an ESP system100that is configured to facilitate anchor shortening across streaming nodes108a-nin accordance with the present disclosure. In response to one or more queries102(which may be provided as input from a user), the ESP system100may ingest one or more event streams116from one or more inputs104, process the event stream(s)116in accordance with the quer(ies)102, and generate output106.

The ESP system100may include a plurality of streaming nodes108a-n, including a first streaming node108a, a second streaming node108b, and an Nth streaming node108n. Each of the streaming nodes108a-nmay be configured to perform one or more processing operations, and the streaming nodes108a-nmay be interconnected to implement the desired processing of the event stream(s)116ingested from the input(s)104. For simplicity, the streaming nodes108a-nare shown as being connected serially. However, the structure of the ESP system100shown inFIG. 1is provided for purposes of example only, and should not be interpreted as limiting the scope of the techniques disclosed herein. More complex topologies are possible in accordance with the present disclosure.

FIG. 1also shows some of the components that may be included in the second streaming node108b, including a downstream marshaller110, a plurality of compute processors112a-n, and an upstream marshaller114. The downstream marshaller110may be configured to pull data from an upstream marshaller (not shown) in the first streaming node108a. The compute processors112a-nmay each be configured to perform one or more processing operations, and they may be interconnected in various ways to implement the desired processing of the event stream(s)116. For simplicity, the compute processors112a-nare shown as being connected serially. Again, however, more complex topologies are possible in accordance with the present disclosure.

The ESP system100may use anchors to facilitate recovery from node failure without loss or duplication of results. Anchors may be passed between various entities within the ESP system100. For example, anchors may be passed between nodes (e.g., from the first streaming node108ato the second streaming node108b). Anchors may also be passed between compute processors within a single node (e.g., from the first compute processor112ato the second compute processor112b).

The size of an anchor may grow as it is passed along a chain of multiple processing entities, such as the chain of compute processors112a-n.FIG. 1shows an anchor120being passed from the Nth compute processor112nto the upstream marshaller114.

Because the anchor120may be generated at the end of a chain of compute processors112a-n, the anchor120may be fairly large and complex. The upstream marshaller114may be configured to map the (potentially complex) anchor120that it receives to a simple index122, such as a single numeric value. The anchor120that the upstream marshaller114receives from the Nth compute processor112nmay be referred to as a full anchor120, and the index122to which the full anchor120is mapped may be referred to herein as an index anchor122. As used herein, the term “full anchor” refers to a complete representation of an anchor that has been mapped to a simpler representation, such as an index. The term “index anchor” refers to an index to which a full anchor has been mapped. The upstream marshaller114may pass the index anchor122to a downstream marshaller in a downstream node, instead of passing the full anchor120itself. In this way, the complexity of anchors within the ESP system100may be limited.

The upstream marshaller114may store mapping information118indicating the relationship between the full anchor120and the index anchor122. If the upstream marshaller114subsequently receives a data request that references the index anchor122(in the event that a downstream node fails, for example), the upstream marshaller114may use the mapping information118to identify the full anchor120that corresponds to the index anchor122.

An example will be discussed in relation toFIGS. 2A-B. Reference is initially made toFIG. 2A, which illustrates how an upstream marshaller214within an ESP system200may map a full anchor to an index anchor. The ESP system200may be configured so that downstream entities pull data from upstream entities. For example, the upstream marshaller214may be configured to pull a data batch226afrom an upstream compute processor212by sending a request224afor the data batch226ato the upstream compute processor212. (In this context, the term “upstream compute processor” refers to a compute processor that is upstream from another processing entity. InFIGS. 2A-B, the upstream compute processor212is upstream from the upstream marshaller214.) The request224amay reference a particular anchor (e.g., AnchorA), which may indicate a point in one or more event streams116where the data batch226ashould begin.

In response to the request224a, the upstream compute processor212may provide the requested data batch226a. In addition to providing the requested data batch226a, the upstream compute processor212may also provide the upstream marshaller214with a new anchor (e.g., AnchorD). The new anchor may indicate one or more end points corresponding to the data batch226a. When the upstream marshaller214makes a subsequent request (not shown) for data from the upstream compute processor212, the subsequent request may reference this new anchor (AnchorD) to indicate where the newly requested data should begin.

The upstream compute processor212may be at the end of a long chain of processing entities (such as the chain of compute processors112a-nshown inFIG. 1), and as a result AnchorD may be fairly complex. For example, AnchorD may be a nested anchor that includes a reference to one or more other anchors. To simplify downstream processing, the upstream marshaller214may map AnchorD to a simple index, which may be a single numeric value (e.g., 3). The upstream marshaller214may also store mapping information indicating the relationship between AnchorD and the corresponding index anchor in a table218. The table218may include multiple records228. Each record228may include a stored index anchor (e.g., 3) and a corresponding stored full anchor (e.g., AnchorD).

An anchor protocol may be designed so that different anchors created by the same processing entity may be compared in order to determine their relative order. For instance, different anchors created by the upstream compute processor212(e.g., AnchorA and AnchorD) may be compared to determine which has a greater value. As a result of this comparison, a determination may be made about the relative order of the anchors being compared. For example, a determination may be made that AnchorD has a greater value than AnchorA. This may indicate that the point in the event stream(s)116corresponding to AnchorD occurs after the point in the event stream(s)116corresponding to AnchorA.

The ESP system200may be configured so that when an upstream marshaller214creates index anchors, the ability to compare different full anchors is preserved. In other words, the ESP system200may be configured so that the value of a first index anchor is greater than the value of a second index anchor if and only if the value of the full anchor corresponding to the first index anchor is greater than the value of the full anchor corresponding to the second index anchor. Thus, the properties of full anchors that permit the full anchors to be compared with one another may be preserved in the index anchors that are created.

To preserve the ability to compare different anchors, the upstream marshaller214may be configured to create new index anchors in sequential order. Referring again to the example depicted inFIG. 2A, suppose that AnchorB has a greater value than AnchorA, AnchorC has a greater value than AnchorB, and so forth. By creating new index anchors in sequential order, the index anchors preserve the relative values of the full anchors. In other words, the index anchor corresponding to AnchorB (which is 1) has a greater value than the index anchor corresponding to AnchorA (which is 0). Similarly, the index anchor corresponding to AnchorC (which is 2) has a greater value than the index anchor corresponding to AnchorB (which is 1), and so forth.

An ESP system in accordance with the present disclosure may be configured so that the ability to compare anchors is preserved even if anchor mapping is performed multiple times. For example, referring briefly to the ESP system100shown inFIG. 1, suppose that upstream marshallers in multiple streaming nodes108a-nperform anchor mapping as disclosed herein (e.g., the index anchor122created by the upstream marshaller114is passed to a subsequent streaming node and additional mapping is performed by a subsequent upstream marshaller). In accordance with the present disclosure, the ability to compare anchors may be preserved even in this type of scenario. Thus, an ESP system may be configured so that anchor mapping as disclosed herein does not affect the ability to compare anchors.

Referring again to the example shown inFIG. 2A, in order to determine the index anchor to which AnchorD should be mapped, the upstream marshaller214may determine the index anchor that was issued most recently. This information may be determined from the table218. Then, the upstream marshaller214may increment the value of the most recently issued index anchor. In the depicted example, it will be assumed that the most recently issued index anchor is the number2. Thus, in this example the upstream marshaller214may determine the index anchor to which AnchorD should be mapped by incrementing the number2, thereby obtaining the number3.

At some subsequent point in time, a downstream marshaller210in a downstream node208bmay pull a data batch226bfrom the upstream marshaller214by sending a request224bto the upstream marshaller214. The request224bmay reference a particular index anchor (e.g., 0), which may indicate a point in one or more event streams116where the data batch226bshould begin.

In response to the request224b, the upstream marshaller214may provide the requested data batch226balong with a new index anchor (e.g., 3). The new index anchor may correspond to the most recent full anchor (AnchorD) that the upstream marshaller214has received from the upstream compute processor212. However, instead of providing the full anchor (AnchorD) along with the requested data batch226b, the upstream marshaller214may instead provide the index anchor (3) to which the full anchor has been mapped. In other words, the upstream marshaller214may pass the index anchor as a new anchor to the downstream marshaller210.

Reference is now made toFIG. 2B, which illustrates how the upstream marshaller214may remap the index anchor (3) back to the full anchor (AnchorD) under some circumstances. If, for example, the downstream node208b(or another node that is farther downstream) fails, the downstream marshaller210may send a request224cfor the upstream marshaller214to resend data that the upstream marshaller214previously sent. The request224cmay include a reference to the index anchor (3) indicating the point in one or more event streams116where the upstream marshaller214should begin resending the data.

In response to receiving the request224c, the upstream marshaller214may use the mapping information that is stored in the table218to determine the full anchor (AnchorD) that corresponds to the index anchor (3) in the request224c. The upstream marshaller214may then pass the full anchor (AnchorD) to the upstream compute processor212by including the full anchor in a request224dthat the upstream marshaller214sends to the upstream compute processor212.

In some implementations, the upstream marshaller214may be configured so that it determines whether the data requested by the downstream marshaller210is stored locally before requesting the data from the upstream compute processor212. For example, the upstream marshaller214may determine whether the requested data is stored in local memory230. If the requested data is not stored in local memory230, the upstream marshaller214may determine whether the requested data is stored in local storage232. If the requested data is not stored in local storage232, the upstream marshaller214may send the request224dfor the data to the upstream compute processor212. Thus, the upstream marshaller214may be configured so that it passes the full anchor (AnchorD) to the upstream compute processor212in response to determining that the data requested by the downstream marshaller210is not stored locally.

The local memory230and the local storage232may be part of the upstream node208a. For example, the upstream node208amay be a computing device, and the local memory230and the local storage232may be part of that computing device. In other words, it may not be necessary to use a network interface to access the local memory230and the local storage232.

FIG. 3illustrates how the mapping techniques disclosed herein may limit the complexity of anchors within an ESP system. Anchors a1through aNare shown growing in complexity as they move down a chain of compute processors312a-nin a streaming node308. An upstream marshaller314may then map the full anchor aN(which is a nested anchor that includes references to anchors a1. . . aN-1) to a simple index anchor such as a single numeric value (e.g., 7).

More specifically, a downstream marshaller310within the streaming node308may pull data from an upstream node (not shown). When the first compute processor312apulls the data from the downstream marshaller310, the downstream marshaller310may provide an anchor a1to the first compute processor312a. When a second compute processor (not shown) pulls data from the first compute processor312a, the first compute processor312amay provide an anchor a2to the second compute processor. The anchor a2may include a reference to the anchor a1that the first compute processor312areceived from the downstream marshaller310.

This pattern of creating a new anchor that references the previously received anchor(s) may continue, such that the anchors grow in complexity as they move down the chain of compute processors312a-n.FIG. 3shows the Nth compute processor312nproviding an anchor aNto the upstream marshaller314. Anchor aNincludes a reference to anchors a1through aN-1.

To simplify downstream processing, the upstream marshaller314may map the full anchor aNto a simple index anchor, which may be a single numeric value (e.g., 7). The upstream marshaller314may also store mapping information in a table318indicating the relationship between the full anchor aNand the corresponding index anchor.

Although the examples of ESP systems discussed above have been fairly simple, more complex topologies are possible.FIGS. 4 and 5show some additional aspects of topologies that may be utilized by an ESP system.

FIG. 4shows an upstream marshaller414connected to multiple downstream marshallers410a-n. Each of the downstream marshallers410a-nmay be configured to pull data from the upstream marshaller414. When the downstream marshallers410a-nrequest data from the upstream marshaller414, the upstream marshaller414may provide an index anchor corresponding to a more complex full anchor, in the manner discussed above.

FIG. 4shows the upstream marshaller414on an upstream node408aand the downstream marshallers410a-non a single downstream node408b. Alternatively, the downstream marshallers410a-nmay be implemented on different nodes.

FIG. 5shows a streaming node508that includes multiple downstream marshallers510a-b, compute processors512a-b, and upstream marshallers514a-b. The streaming node508also includes an input processor534.

The downstream marshallers510a-bmay be configured to pull data from an upstream marshaller in an upstream node (not shown). The first compute processor512amay be configured to pull data from the downstream marshallers510a-b. The first upstream marshaller514amay be configured to pull data from a first compute processor512a. A downstream marshaller in a downstream node (not shown) may be configured to pull data from the first upstream marshaller514a.

The input processor534may be configured to pull data from an input (not shown). The second compute processor512bmay be configured to pull data from the input processor534. The second upstream marshaller514bmay be configured to pull data from the second compute processor512b.

When data is passed from one entity to another, an anchor may be provided. Each of the upstream marshallers514a-bmay be configured to map a full anchor to an index anchor, in the manner discussed above.

FIG. 6illustrates an example of a method600for facilitating anchor shortening across streaming nodes in an ESP system. The method600will be described in relation to the ESP system200shown inFIGS. 2A-B. The method600may be implemented by an upstream marshaller214within the ESP system200.

The method600may include sending602a request224afor a data batch226ato an upstream compute processor212. The request224amay include a reference to a particular anchor that indicates the point in one or more event streams116where the requested data batch226ashould begin. In response to the request224a, the upstream marshaller214may receive604the requested data batch226aalong with a new anchor.

To simplify downstream processing, the upstream marshaller214may map606the full anchor received from the upstream compute processor212to an index anchor, which may be a single numeric value. The upstream marshaller214may also store608mapping information indicating the relationship between the full anchor and the index anchor. The mapping information may be stored608in a table218, for example. At some subsequent point in time, the upstream marshaller214may pass610the index anchor as a new anchor to a downstream entity, such as a downstream marshaller210. For example, when a downstream marshaller210sends a request224bto the upstream marshaller214for a data batch226b, the upstream marshaller214may provide the requested data batch226balong with the index anchor.

FIG. 7illustrates an example of a method700for remapping in an ESP system that implements an anchor protocol. The method700will be described in relation to the ESP system200shown inFIGS. 2A-B. The method700may be implemented by an upstream marshaller214within the ESP system200.

The upstream marshaller214may receive702, from a downstream entity such as a downstream marshaller210, a request224cto resend data that the upstream marshaller214previously sent. The downstream marshaller210may send such a request224cin response to failure of a downstream node (such as the downstream node208bthat includes the downstream marshaller210, or another node that is farther downstream). The request224cmay include a reference to an index anchor that the upstream marshaller214previously provided to the downstream marshaller210.

In response to receiving702the request224c, the upstream marshaller214may determine704whether the requested data is stored locally. If it is, the upstream marshaller214may provide706the requested data.

If, however, the requested data is not stored locally, the upstream marshaller214may use mapping information to determine708the full anchor that corresponds to the index anchor in the request224c. In other words, the upstream marshaller214may remap the index anchor to the full anchor. The upstream marshaller214may then pass710the full anchor to the upstream compute processor212by including the full anchor in a request224dthat the upstream marshaller214sends to the upstream compute processor212.

FIG. 8illustrates certain components that may be included within a computer system800. One or more computer systems800may be used to implement the various devices, components, and systems described herein, including the ESP systems100,200and the streaming nodes108a-n,208a-b,308,408,508.

The computer system800includes a processor801. The processor801may be a general purpose single- or multi-chip microprocessor (e.g., an Advanced RISC (Reduced Instruction Set Computer) Machine (ARM)), a special purpose microprocessor (e.g., a digital signal processor (DSP)), a microcontroller, a programmable gate array, etc. The processor801may be referred to as a central processing unit (CPU). Although just a single processor801is shown in the computer system800ofFIG. 8, in an alternative configuration, a combination of processors (e.g., an ARM and DSP) could be used.

The computer system800also includes memory803. The memory803may be any electronic component capable of storing electronic information. For example, the memory803may be embodied as random access memory (RAM), read-only memory (ROM), magnetic disk storage media, optical storage media, flash memory devices in RAM, on-board memory included with the processor, erasable programmable read-only memory (EPROM), electrically erasable programmable read-only memory (EEPROM) memory, registers, and so forth, including combinations thereof.

Instructions805and data807may be stored in the memory803. The instructions805may be executable by the processor801to implement some or all of the methods disclosed herein, such as the methods600,700shown inFIGS. 6 and 7. Executing the instructions805may involve the use of the data807that is stored in the memory803. Any of the various examples of modules and components described herein may be implemented, partially or wholly, as instructions805stored in memory803and executed by the processor801. Any of the various examples of data described herein may be among the data807that is stored in memory803and used during execution of the instructions805by the processor801.

A computer system800may also include one or more communication interfaces809for communicating with other electronic devices. The communication interface(s)809may be based on wired communication technology, wireless communication technology, or both. Some examples of communication interfaces809include a Universal Serial Bus (USB), an Ethernet adapter, a wireless adapter that operates in accordance with an Institute of Electrical and Electronics Engineers (IEEE) 802.11 wireless communication protocol, a Bluetooth wireless communication adapter, and an infrared (IR) communication port.

A computer system800may also include one or more input devices811and one or more output devices813. Some examples of input devices811include a keyboard, mouse, microphone, remote control device, button, joystick, trackball, touchpad, and lightpen. Some examples of output devices813include a speaker and a printer. One specific type of output device that is typically included in a computer system800is a display device815. Display devices815used with embodiments disclosed herein may utilize any suitable image projection technology, such as liquid crystal display (LCD), light-emitting diode (LED), gas plasma, electroluminescence, or the like. A display controller817may also be provided, for converting data807stored in the memory803into text, graphics, and/or moving images (as appropriate) shown on the display device815.

In accordance with an aspect of the present disclosure, a method is disclosed for facilitating anchor shortening across streaming nodes in an event stream processing system. The method may include receiving a full anchor at an upstream marshaller. The full anchor may be associated with a data batch that corresponds to one or more event streams. The full anchor may include an indication of an input point for the one or more event streams. The full anchor may be received from an upstream compute processor. The method may also include mapping the full anchor to an index anchor and passing the index anchor to a downstream marshaller.

The full anchor may be a nested anchor that includes a reference to at least one other anchor.

The method may further include creating index anchors in sequential order, such that properties of full anchors that permit the full anchors to be compared with one another are preserved in the index anchors.

The method may further include storing mapping information indicating a relationship between the full anchor and the index anchor. The mapping information may be stored in a table having a plurality of records. Each record may include a stored index anchor and a corresponding stored full anchor.

The full anchor may be received at the upstream marshaller in response to a data request from the upstream marshaller. The method may further include receiving a requested data batch at the upstream marshaller.

The index anchor may be passed to the downstream marshaller in response to a data request from the downstream marshaller. The method may further include sending requested data to the downstream marshaller with the index anchor.

An upstream node may include the upstream marshaller. A downstream node may include the downstream marshaller. The upstream node and the downstream node may include separate computing devices.

In accordance with another aspect of the present disclosure, a method is disclosed for remapping in an event stream processing system that implements an anchor protocol. The method may include receiving an index anchor at an upstream marshaller. The index anchor may be received from a downstream marshaller. The index anchor may correspond to a full anchor that the upstream marshaller previously received from an upstream compute processor. The full anchor may be associated with a data batch that corresponds to one or more event streams. The full anchor may include an indication of an input point for the one or more event streams. The method may include using mapping information to determine the full anchor that corresponds to the index anchor, and passing the full anchor to the upstream compute processor.

The index anchor may be received in response to failure of a downstream node.

The full anchor may be a nested anchor that includes a reference to at least one other anchor.

The index anchor may be received at the upstream marshaller with a data request. The method may further include determining whether a requested data batch is stored locally. The full anchor may be passed to the upstream compute processor in response to determining that the requested data batch is not stored locally.

An upstream node may include the upstream marshaller. A downstream node may include the downstream marshaller. The upstream node and the downstream node may include separate computing devices.

In accordance with another aspect of the present disclosure, an event stream processing system may be configured to facilitate anchor shortening across streaming nodes. The event stream processing system may include a downstream marshaller, an upstream compute processor, and an upstream marshaller that is configured to receive a full anchor from the upstream compute processor. The full anchor may be associated with a data batch that corresponds to one or more event streams. The full anchor may include an indication of an input point for the one or more event streams. The upstream marshaller may be additionally configured to map the full anchor to an index anchor and pass the index anchor to the downstream marshaller.

The full anchor may be a nested anchor that includes a reference to at least one other anchor.

The upstream marshaller may be additionally configured to receive the index anchor from the downstream marshaller, use mapping information to determine the full anchor that corresponds to the index anchor, and pass the full anchor to the upstream compute processor.

The upstream marshaller may receive the index anchor with a data request. The upstream marshaller may be additionally configured to determine whether a requested data batch is stored locally. The upstream marshaller may pass the full anchor to the upstream compute processor in response to determining that the requested data batch is not stored locally.

The upstream marshaller may be configured to create index anchors in sequential order, such that properties of full anchors that permit the full anchors to be compared with one another are preserved in the index anchors.

The upstream marshaller may be additionally configured to store mapping information indicating a relationship between the full anchor and the index anchor.

The upstream marshaller may receive the full anchor from the upstream compute processor in response to a data request from the upstream marshaller. The upstream marshaller may be additionally configured to receive a requested data batch with the full anchor.

The upstream marshaller may pass the index anchor to the downstream marshaller in response to a data request from the downstream marshaller. The upstream marshaller may be additionally configured to send requested data to the downstream marshaller with the index anchor.