# US 9,973,558 B2 · Distributed streaming platform for real-time applications > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 9,973,558 B2 - Document ID: US9973558B2 - Status: Granted - Kind: B2 - Application number: 15/614,260 - Filed: 2017-06-05 - Published: 2017-11-23 - Granted: 2018-05-15 - Earliest claimed: 2013-03-11 - Assignees: DataTorrent, Inc. - Applicants: DataTorrent, Inc. - Inventors: Amol J. Kekre; Phu D. Hoang; Chetan Narsude; Thomas Weise - Web page: https://patentgrep.com/patent/US9973558B2 ## CPC - H04L 65/60: Network streaming of media packets - G06F 16/24568: {Data stream processing; Continuous queries} - H04L 65/612: for unicast - H04L 65/80: Responding to QoS - H04L: TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION (arrangements common to telegraphic and telephonic communication H04M) - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract Software for a distributed streaming platform receives a specification for a streaming application and converts the specification into a logical plan that includes a directed acyclic graph (DAG) with operators as nodes and streams as edges. The software translates the logical plan into a physical plan using any stream modes as specified and with one or more instances of the operators per any static partitioning. Then the software obtains a plurality of containers running on a networked cluster of servers. One slave executes in each container. The software assigns the instances to the slaves for execution according to an execution plan that depends at least in part on the physical plan and number of containers obtained. Each slave monitors throughput of the streaming windows through the instances in the container and other statistics related to service level and reports the results of the monitoring, which are used for dynamic adjustment. ## Claims ### Claim 1 A method, comprising the operations of: receiving a specification for a streaming application and converting the specification into a logical plan that includes a directed acyclic graph (DAG) with operators as nodes and streams as edges, wherein the operators are program instructions and the streams are unbound sequences of streaming windows that are ordered in terms of time; translating the logical plan into a physical plan using stream modes specified in the specification and with one or more instances of the operators per any static partitioning in the specification, wherein at least one of the stream modes is in-rack and at least one of the stream modes is in-node; obtaining a plurality of containers running on a networked cluster of servers, wherein one slave executes in each container, wherein each container includes a buffer server that manages a message queue for the streaming windows; assigning the instances to the slaves for execution according to an execution plan that depends at least in part on the physical plan and number of containers obtained, wherein each slave monitors throughput of the streaming windows through the instances in the corresponding container and other statistics related to service level and reports results of the monitoring; launching the execution plan by converting external input into tuples grouped into streaming windows; determining an instance of a first operator of a first container is creating a bottleneck based on the throughput monitoring by each slave; creating multiple instances of the first operator and assigning the multiple instances to the first container; hashing each tuple of a first streaming window for the first operator using a hash function; assigning each tuple of the first streaming window for the first operator to one of the multiple instances based on selected bits of a corresponding hash code generated by the hashing; making one or more dynamic adjustments based on the results of the monitoring including re-initiating the streams in accordance with a recovery policy; and outputting the tuples of one or more instances for display and to persistent storage for subsequent use. ### Claim 2 (depends on 1) The method of claim 1, wherein the assigning further comprises: assigning each tuple based on lower bits of the corresponding hash code. ### Claim 3 (depends on 1) The method of claim 1, further comprising: determining a skewed distribution of tuples among the multiple instances of the first operator; and merging two instances of the multiple instances into a combined instance. ### Claim 4 (depends on 1) The method of claim 1, further comprising; determining a skewed distribution of tuples among the multiple instances of the first operator; and splitting a single instance of the multiple instances into two or more instances. ### Claim 5 (depends on 4) The method of claim 4, further comprising: merging the streams from the two or more instances downstream. ### Claim 6 (depends on 1) The method of claim 1, wherein each container is a multi-thread process including one thread for each operator instance. ### Claim 7 (depends on 1) The method of claim 1, wherein each stream inputs a sequence of sequential structured tuples to an operator or outputs a sequence of sequential structured tuples from an operator. ### Claim 8 (depends on 1) The method of claim 1, wherein each operator executes as a single thread in a multi-thread process that also includes at least one thread for the slave and at least one thread for a buffer server. ### Claim 9 A non-transitory computer-readable medium storing a computer program for implementing a method, the computer-readable medium comprising: program instructions for receiving a specification for a streaming application and converting the specification into a logical plan that includes a directed acyclic graph (DAG) with operators as nodes and streams as edges, wherein the operators are program instructions and the streams are unbound sequences of streaming windows that are ordered in terms of time; program instructions for translating the logical plan into a physical plan using stream modes specified in the specification and with one or more instances of the operators per any static partitioning in the specification, wherein at least one of the stream modes is in-rack and at least one of the stream modes is in-node; program instructions for obtaining a plurality of containers running on a networked cluster of servers, wherein one slave executes in each container, wherein each container includes a buffer server that manages a message queue for the streaming windows; program instructions for assigning the instances to the slaves for execution according to an execution plan that depends at least in part on the physical plan and number of containers obtained, wherein each slave monitors throughput of the streaming windows through the instances in the corresponding container and other statistics related to service level and reports results of the monitoring; program instructions for launching the execution plan by converting external input into tuples grouped into streaming windows; program instructions for determining an instance of a first operator of a first container is creating a bottleneck based on the throughput monitoring by each slave; program instructions for creating multiple instances of the first operator and assigning the multiple instances to the first container; program instructions for hashing each tuple of a first streaming window for the first operator using a hash function; program instructions for assigning each tuple of the first streaming window for the first operator to one of the multiple instances based on selected bits of a corresponding hash code generated by the hashing; program instructions for making one or more dynamic adjustments based on the results of the monitoring including re-initiating the streams in accordance with a recovery policy; and program instructions for outputting the tuples of one or more instances for display and to persistent storage for subsequent use. ### Claim 10 (depends on 9) The computer-readable medium of claim 9, wherein the program instructions for assigning further comprises: program instructions for assigning each tuple based on lower bits of the corresponding hash code. ### Claim 11 (depends on 9) The computer-readable medium of claim 9, further comprising: program instructions for determining a skewed distribution of tuples among the multiple instances of the first operator; and program instructions for merging two instances of the multiple instances into a combined instance. ### Claim 12 (depends on 9) The computer-readable medium of claim 9, further comprising; program instructions for determining a skewed distribution of tuples among the multiple instances of the first operator; and program instructions for splitting a single instance of the multiple instances into two or more instances. ### Claim 13 (depends on 12) The computer-readable medium of claim 12, further comprising: program instructions for merging the streams from the two or more instances downstream. ### Claim 14 (depends on 9) The computer-readable medium of claim 9, wherein in the program instructions each container is a multi-thread process including one thread for each operator instance. ### Claim 15 (depends on 9) The computer-readable medium of claim 9, wherein in the program instructions each stream inputs a sequence of sequential structured tuples to an operator or outputs a sequence of sequential structured tuples from an operator. ### Claim 16 (depends on 9) The computer-readable medium of claim 9, wherein in the program instructions each operator executes as a single thread in a multi-thread process that also includes at least one thread for the slave and at least one thread for a buffer server. ### Claim 17 A computer system comprising: a processor; and memory coupled to the processor and having stored therein instructions that, if executed by the computer system, cause the computer system to execute a method comprising: receiving a specification for a streaming application and converting the specification into a logical plan that includes a directed acyclic graph (DAG) with operators as nodes and streams as edges, wherein the operators are program instructions and the streams are unbound sequences of streaming windows that are ordered in terms of time; translating the logical plan into a physical plan using stream modes specified in the specification and with one or more instances of the operators per any static partitioning in the specification, wherein at least one of the stream modes is in-rack and at least one of the stream modes is in-node; obtaining a plurality of containers running on a networked cluster of servers, wherein one slave executes in each container, wherein each container includes a buffer server that manages a message queue for the streaming windows; assigning the instances to the slaves for execution according to an execution plan that depends at least in part on the physical plan and number of containers obtained, wherein each slave monitors throughput of the streaming windows through the instances in the corresponding container and other statistics related to service level and reports results of the monitoring; launching the execution plan by converting external input into tuples grouped into streaming windows; determining an instance of a first operator of a first container is creating a bottleneck based on the throughput monitoring by each slave; creating multiple instances of the first operator and assigning the multiple instances to the first container; hashing each tuple of a first streaming window for the first operator using a hash function; assigning each tuple of the first streaming window for the first operator to one of the multiple instances based on selected bits of a corresponding hash code generated by the hashing; making one or more dynamic adjustments based on the results of the monitoring including re-initiating the streams in accordance with a recovery policy; and outputting the tuples of one or more instances for display and to persistent storage for subsequent use. ### Claim 18 (depends on 17) The computer system of claim 17, wherein the method further comprises: determining a skewed distribution of tuples among the multiple instances of the first operator; and merging two instances of the multiple instances into a combined instance. ### Claim 19 (depends on 17) The computer system of claim 17, wherein the method further comprises; determining a skewed distribution of tuples among the multiple instances of the first operator; and splitting a single instance of the multiple instances into two or more instances. ### Claim 20 (depends on 19) The computer system of claim 19, wherein the method further comprises: merging the streams from the two or more instances downstream.