PARTITION-BASED DATA STREAM PROCESSING FRAMEWORK

• US 20150134626A1
• Filed: 11/11/2013
• Published: 05/14/2015
• Est. Priority Date: 11/11/2013
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: THEIMER, MARVIN MICHAEL, GHARE, GAURAV D., DUNAGAN, JOHN DAVID, BURGESS, GREG, XIONG, YING
• Granted Patent: US 10,635,644 B2
• Time in Patent Office: Days
• US Class Current: 707/693
• CPC Class Codes: G06F 11/1469 Backup restoration techniques
• CPC Class Codes: G06F 11/1482 by means of middleware or OS functionality
• CPC Class Codes: G06F 11/2028 eliminating a faulty processor or activating a spare
• CPC Class Codes: G06F 11/3006 where the computing system is distributed, e.g. networked systems, clusters, multiprocessor systems multiprogramming arrangements G06F9/46; allocation of resources G06F9/50
• CPC Class Codes: G06F 11/3055 Monitoring arrangements for monitoring the status of the computing system or of the computing system component, e.g. monitoring if the computing system is on, off, available, not available error or fault processing without redundancy G06F11/0703; error detection or correction by redundancy in data representation G06F11/08; error detection or correction by redundancy in operation G06F11/14; error detection or correction by redundancy in hardware G06F11/16
• CPC Class Codes: G06F 11/3433 for load management allocation of a server based on load conditions G06F9/505; load rebalancing G06F9/5083; redistributing the load in a network by a load balancer H04L67/1029
• CPC Class Codes: G06F 16/21 Design, administration or maintenance of databases
• CPC Class Codes: G06F 9/5066 Algorithms for mapping a plurality of inter-dependent sub-tasks onto a plurality of physical CPUs mappping at compile time, see G06F8/451

Abstract

A control node of a multi-tenant stream processing service receives a request indicating an operation to be performed on data records of a particular data stream. Based on a stream partitioning policy, the control node determines an initial number of worker nodes to be used. The control node configures a worker node to perform the operation on received records. In response to a determination that the worker node is in an unhealthy state, the control node configures a replacement worker node.

Claims

1. A system, comprising:

one or more computing devices configured to;

implement one or more programmatic interfaces enabling a client of a multi-tenant stream processing service to indicate, corresponding to a particular processing stage associated with a specified data stream;

(a) a processing operation to be performed on data records of the specified data stream in accordance with a partitioning policy and (b) an output distribution descriptor for results of the processing operations;

receive, from a particular client via the one or more programmatic interfaces, an indication of a particular processing operation to be performed on data records of a particular data stream at the particular processing stage, and a particular output distribution descriptor for results of the particular processing operation;

determine, based at least in part on the partitioning policy and at least in part on an estimated performance capability of resources to be deployed as worker nodes for the processing stage, an initial number of worker nodes for the specified data stream;

configure a particular worker node of the initial number of worker nodes to (a) receive data records of one or more partitions of the particular data stream, (b) perform the particular processing operation on received data records, (c) store progress records indicating the portion of the one or more partitions that have been processed at the worker node, and (d) transfer results of the particular processing operations to one or more destinations in accordance with the particular output distribution descriptor;

monitor a health state of the particular worker node; and

in response to a determination that the particular worker node is in an undesired state, configure a replacement worker node to replace the particular worker node, wherein the replacement worker node accesses a progress record stored by the particular worker node to identify at least one data record of the one or more partitions on which the particular processing operation is to be performed by the replacement worker node.

2. The system as recited in claim 1, wherein the particular output distribution descriptor indicates that results of the particular processing operations are to be distributed, as data records of a different data stream, to or more ingestion nodes configured for the different data stream, in accordance with a different partitioning policy.

3. The system as recited in claim 1, wherein the one or more computing devices are further configured to:

receive, from the particular client, an indication of another processing stage to which data records of the particular data stream are to be provided as input, wherein a different processing operation is to be performed at the other processing stage; and

configure an additional set of worker nodes for the other processing stage.

4. The system as recited in claim 1, wherein the one or more computing devices is further configured to:

in response to a determination that a different worker node configured to perform another processing operation for another processing stage is in an undesired state, configure a different replacement worker node to perform the other processing operation on one or more subsequently-received data records without accessing a progress record.

5. The system as recited in claim 1, wherein the one or more computing devices is further configured to:

in response to a determination that a workload level at a different worker node of the processing stage meets a triggering criterion, implement a stage reconfiguration operation comprising one or more of;

(a) a dynamic repartitioning of the particular data stream performed while additional data records of the stream continue to be processed, (b) an assignment of an alternate worker node to at least one partition previously being processed at the different worker node, (c) a change to a number of worker nodes configured for the processing stage, or (d) a transfer of a worker node from one server to another server.

6. A method, comprising:

performing, by one or more computing devices;

receiving, at a multi-tenant stream processing service from a particular client, an indication of a particular operation to be performed on data records of a particular data stream at a specified processing stage, and a particular output distribution descriptor for results of the particular operation;

determining, based at least in part on the particular operation, an initial number of worker nodes to be configured for the specified processing stage;

configuring a particular worker node of the initial number of worker nodes to (a) perform the particular operation on received data records of one or more partitions of the particular data stream, (b) store progress records indicating the portion of the one or more partitions that have been processed at the worker node, and (c) transfer results of the particular operations to one or more destinations in accordance with the particular output distribution descriptor; and

in response to a determination that the particular worker node is in an unhealthy state, selecting a replacement worker node to replace the particular worker node, wherein the replacement worker node accesses a progress record stored by the particular worker node to identify at least one data record of the one or more partitions on which the particular operation is to be performed by the replacement worker node.

7. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

invoking one or more programmatic data record retrieval interfaces implemented by a multi-tenant stream management service to receive data records of the one or more partitions, including a particular programmatic data record retrieval interface that includes, as a parameter, an indication of a sequence number within a partition of a requested data record.

8. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

implementing one or more programmatic interfaces to enable clients of the stream processing service to specify a directed acyclic graph of processing stages for data records of one or more data streams.

9. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

obtaining, from a multi-tenant stream management service responsible for storage of the data records of the particular data stream, an indication of a partitioning policy being used for the particular data stream; and

determining the initial number of worker nodes based at least in part on the partitioning policy.

10. The method as recited in claim 6, wherein the particular output distribution descriptor indicates that results of the particular operations are to be distributed, as data records of a different data stream, to or more ingestion nodes configured for the different data stream in accordance with a different partitioning policy.

11. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

in response to determining that a different worker node configured to perform another operation for another processing stage is in an undesired state, configuring a different replacement worker node to perform the other operation on one or more subsequently-received data records without accessing a progress record.

12. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

in response to determining that a workload level at a different worker node of the processing stage meets a triggering criterion, implementing one or more of;

(a) a dynamic repartitioning of the particular data stream, (b) an assignment of an alternate worker node to at least one partition previously being processed at the different worker node, (c) a change to a number of worker nodes configured for the processing stage, or (d) a transfer of a worker node from one server to another server.

13. The method as recited in claim 6, wherein a result of the particular operation is formatted in accordance with an input format compatible with another stream processing system, and wherein a particular destination of the one or more destinations comprises an input node of the other stream processing system.

14. The method as recited in claim 6 wherein the particular worker node is configured to store, in a persistent data repository, an entry representing accumulated application state information corresponding to a plurality of data records that have been processed at the particular worker node, and to include an indication of the entry in a progress record.

15. The method as recited in claim 6, wherein the operation comprises one of:

a log record analysis operation, a resource monitoring metric analysis, a billing amount computation, a sensor data analysis, a social media interaction analysis, a real-time ETL (Extract-Transform-Load) processing operation, or a transformation of a data record prior to its insertion into a data warehouse.

16. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

in response to a stream processing configuration request via an invocation of a client library component, registering, at the multi-tenant stream processing service, a specified resource as a worker node for a different processing stage.

17. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

in response to a stream processing configuration request via an invocation of a client library component, determining, at the multi-tenant stream processing service, one or more categories of control-plane functions to be implemented at a different processing stage.

18. The method as recited in claim 6, wherein the operation is an idempotent operation.

19. The method as recited in claim 6, further comprising performing, by the one or more computing devices:

receiving, at the multi-tenant stream processing service from the particular client, an indication of a particular non-idempotent operation to be performed on data records of a different data stream at a different processing stage; and

configuring a first worker node of the different processing stage to perform the non-idempotent operation on received data records.

20. The method as recited in claim 19, further comprising performing, by the one or more computing devices:

configuring the first worker node of the different processing stage to (a) perform a flush operation to store results of the non-idempotent operations to one or more destinations, and (b) save an indication of a flush operating timing at a persistent storage location; and

configuring a replacement worker node to replay, using the indication of the flush operation timing, the flush operation during recovery after a failure of the first worker node.

21. A non-transitory computer-accessible storage medium storing program instructions that when executed on one or more processors implement a control node of a multi-tenant stream processing service, wherein the control node is operable to:

receive, from a particular client via a programmatic interface, an indication of a particular operation to be performed on data records of a particular data stream;

determine, based at least in part on a partitioning policy associated with the particular data stream, an initial number of worker nodes for the specified data stream at a processing stage;

configure a particular worker node of the initial number of worker nodes to perform the particular operation on received data records of one or more partitions of the particular data stream; and

in response to a determination that the particular worker node is in an unhealthy state, configure a replacement worker node to replace the particular worker node.

22. The non-transitory computer-accessible storage medium as recited in claim 21, wherein the control node is operable to:

configure a redundancy group comprising a plurality of worker nodes to process data records of a different partition of a different data stream, wherein at least one worker node of the plurality of worker nodes is designated as a primary node to receive the data records of the different partition, and wherein at least another worker node of the plurality is configured as a standby node to assume responsibilities of a primary node in response to a triggering event.

23. The non-transitory computer-accessible storage medium as recited in claim 21, wherein the control node is operable to:

in response to a determination that a workload level at a different worker node of the processing stage meets a triggering criterion, implementing one or more of;

(a) a dynamic repartitioning of the particular data stream, (b) an assignment of an alternate worker node to at least one partition previously being processed at the different worker node, (c) a change to a number of worker nodes configured for the processing stage, or (d) a transfer of a worker node from one server to another server.

24. The non-transitory computer-accessible storage medium as recited in claim 21, wherein the particular output distribution descriptor indicates that results of the particular operations are to be distributed, as data records of an ephemeral data stream for which storage to a persistent storage device is not a requirement, to or more ingestion nodes configured for the ephemeral data stream in accordance with a different partitioning policy.

25. The non-transitory computer-accessible storage medium as recited in claim 21, wherein the control node is operable to:

implement one or more programmatic interfaces to enable clients of the stream processing service to specify a directed acyclic graph of processing stages for data records of one or more data streams.