# US 11,030,169 B1 · Data re-sharding

> Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt

- Number: US 11,030,169 B1
- Document ID: US11030169B1
- Status: Granted
- Kind: B1
- Application number: 15/452,587
- Filed: 2017-03-07
- Granted: 2021-06-08
- Earliest claimed: 2017-03-07
- Assignees: Amazon Technologies, Inc.
- Applicants: Amazon Technologies, Inc.
- Inventors: Ming-Chuan Wu; Sandeep Bhatia; Andrew Whitaker
- Web page: https://patentgrep.com/patent/US11030169B1

## CPC

- G06F 16/22: Indexing; Data structures therefor; Storage structures
- G06F 16/24554: {Unary operations; Data partitioning operations}
- G06F 16/278: {Data partitioning, e.g. horizontal or vertical partitioning}
- H04L 67/1031: Controlling of the operation of servers by a load balancer, e.g. adding or removing servers that serve requests
- H04L 67/1097: for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]
- G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N)
- H04L: TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION (arrangements common to telegraphic and telephonic communication H04M)

## Abstract

Processing and storage responsibility for a data set may be split according to separately stored shards of the data set. As one or more loads associated with shards of the dataset grow a re-sharding operation may be performed to reduce loading of particular shards and nodes that host the particular shards. A re-sharding operation may cause only a sub-set of as set of shards of the dataset to be split and only cause second portions of the split shards to be stored in additional computing nodes. In some embodiments, a number of shards to be included in the sub-set of shards to be split may be selected based on an overall number of shards in the set and a largest number in the Fibonacci sequence that is less than the overall number of shards in the set.

## Claims

### Claim 1

A system, comprising:
one or more computer systems comprising computer hardware configured to implement a resource manager to:
cause a set of shards of a common dataset to be stored in storages of a first set of computing nodes;
determine a number of new shards to add for the common dataset, wherein the number is determined based on a current number of shards in the set and is less than the current number; and
create the determined number of new shards for the common dataset, wherein create the determined number of new shards for the common dataset comprises:
identify an equivalent number of existing shards from the set of shards as a sub-set of shards to split;
cause only each shard of the sub-set of the set of shards stored in the storages of the first set of computing nodes to be split into a first portion and a second portion; and
cause only the second portions of the first and second portions to be stored in storages of a second set of computing nodes.

### Claim 2 (depends on 1)

The system of claim 1, wherein the number of new shards to add for the common dataset is determined based at least in part on a largest number in a Fibonacci sequence less than the number of shards in the set, wherein the Fibonacci sequence begins with zero or a positive integer.

### Claim 3 (depends on 1)

The system of claim 1, wherein the resource manager causes shards to be included in the sub-set of shards based on respective loads of the computing nodes.

### Claim 4 (depends on 1)

The system of claim 1, wherein said split of the shards of the sub-set load balances a set of shards that includes the sub-set without performing a rebalancing operation subsequent to causing only the second portions to be stored in the storages of the second set of computing nodes.

### Claim 5 (depends on 1)

The system of claim 1, wherein the sub-set of shards comprise non-relational formatted data.

### Claim 6

A method, comprising:
causing a set of shards of a common dataset to be stored in storages of a first set of computing nodes;
determining a number of new shards to add for the common dataset, wherein the number is determined based on a current number of shards in the set and is less than the current number; and
creating the determined number of new shards for the common dataset, comprising:
identifying an equivalent number of existing chards from the set of shards as a sub-set of shards to split;
causing only each shard of the sub-set of the set of shards stored in the storages of the first set of computing nodes to be split into a first portion and a second portion; and
causing only the second portions of the first and second portions to be stored in storages of a second set of computing nodes.

### Claim 7 (depends on 6)

The method of claim 6,
wherein said determining the number of new shards to add for the common dataset is based on a largest number in a Fibonacci sequence less than the number of shards in the set, wherein the Fibonacci sequence begins with zero or a positive integer.

### Claim 8 (depends on 6)

The method of claim 6, further comprising:
performing a split of at least one shard of the sub-set in parallel with performing a split of at least one other shard of the sub-set.

### Claim 9 (depends on 6)

The method of claim 6, further comprising:
causing shards to be included in the sub-set based on respective loads of the computing nodes within which the shards are stored.

### Claim 10 (depends on 9)

The method of claim 9, wherein the respective loads of the computing nodes are based on a data storage rate, a data read rate, an amount of available memory, or a storage capacity of the respective computing nodes.

### Claim 11 (depends on 9)

The method of claim 9, wherein the respective loads of the computing nodes are based on node processor usage or node network usage.

### Claim 12 (depends on 6)

The method of claim 6, further comprising:
causing a directory to be updated to indicate storage locations of data associated with the second portions.

### Claim 13 (depends on 6)

The method of claim 6, wherein said causing the set of shards stored within the sub-set of the first set of computing nodes to be split is performed:
automatically in response to a load associated with the set of shards that includes the sub-set exceeding a threshold amount; or
in response to a command to perform a re-sharding of the set of shards that includes the sub-set.

### Claim 14

A non-transitory, computer-readable storage medium, storing program instructions that when executed by one or more computing devices cause the one or more computing devices to:
cause a set of shards of a common dataset to be stored in storages of a first set of computing nodes;
determine a number of new shards to add for the common dataset, wherein the number is determined based on a current number of shards in the set and is less than the current number; and
create the determined number of new shards for the common dataset, wherein create the determined number of new shards for the common dataset comprises:
identify an equivalent number of existing shards from the set of shards as a sub-set of shards to split;
cause only each shard of the sub-set of shards of the set of shards stored in the storages of the first set of computing nodes to be split into a first portion and a second portion; and
cause only the second portions of the first and second portions to be stored in storages of a second set of computing nodes.

### Claim 15 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein to determine the number of new shards to add for the common dataset, the program instructions further cause the one or more computing devices to determine the number of shards to add for the common dataset based on a largest number in a Fibonacci sequence less than the number of shards in the set, wherein the Fibonacci sequence begins with zero or a positive integer.

### Claim 16 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein the program instructions further cause the one or more computing devices to:
cause splitting of at least two shards of the sub-set to be performed in parallel.

### Claim 17 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein the program instructions further cause the one or more computing devices to:
cause shards to be included in the sub-set based on respective loads of the first set of computing nodes within which the set of shards are stored.

### Claim 18 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein the respective loads of the computing nodes are based on I/O volume or network usage.

### Claim 19 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein the respective loads of the computing nodes are based on, memory capacity, storage capacity, or processor usage of the respective computing nodes.

### Claim 20 (depends on 14)

The non-transitory, computer-readable storage medium of claim 14, wherein the program instructions further cause the one or more computing devices to:
cause a directory to be updated to indicate storage locations of data associated with the second portions.