Systems, methods, and devices for storage shuffle acceleration

A method of processing data in a system having a host and a storage node may include performing a shuffle operation on data stored at the storage node, wherein the shuffle operation may include performing a shuffle write operation, and performing a shuffle read operation, wherein at least a portion of the shuffle operation is performed by an accelerator at the storage node. A method for partitioning data may include sampling, at a device, data from one or more partitions based on a number of samples, transferring the sampled data from the device to a host, determining, at the host, one or more splitters based on the sampled data, communicating the one or more splitters from the host to the device, and partitioning, at the device, data for the one or more partitions based on the one or more splitters.

TECHNICAL FIELD

This disclosure relates generally to data shuffle operations, and more specifically to systems, methods, and apparatus for near-storage shuffle acceleration.

BACKGROUND

A system having a host and one or more storage nodes may utilize a shuffle operation, for example, to rearrange data between partitions and/or nodes.

SUMMARY

A method of processing data in a system having a host and a storage node may include performing a shuffle operation on data stored at the storage node, wherein the shuffle operation may include performing a shuffle write operation, and performing a shuffle read operation, wherein at least a portion of the shuffle operation is performed by an accelerator at the storage node. The portion of the shuffle operation performed at the storage node may include a portion of the shuffle write operation. The portion of the shuffle write operation may include a partition operation. The portion of the shuffle write operation may include one or more of an aggregation operation, a sort operation, a merge operation, a serialize operation, a compression operation, or a spill operation. The portion of the shuffle operation performed at the storage node may include a portion of the shuffle read operation. The portion of the shuffle read operation may include one or more of a fetching operation, a decompression operation, a deserialize operation, a merge operation, a sort operation, or an aggregation operation. The portion of the shuffle operation performed at the storage node may include a partition operation performed using a peer-to-peer (P2P) connection between an accelerator and a storage device at the storage node. The portion of the shuffle operation performed at the storage node may include a data spill operation performed using a P2P connection between an accelerator and a storage device at the storage node. The portion of the shuffle operation performed at the storage node may include a fetch operation performed using a direct memory access operation. The portion of the shuffle operation performed at the storage node may include a data merge operation performed using a P2P connection between an accelerator and a storage device at the storage node.

A storage node may include a storage device, and an accelerator, wherein the storage node is configured to perform at least a portion of a shuffle operation using the accelerator. The storage node may further include a P2P connection between the storage device and the accelerator, and the storage device and the accelerator may be configured to perform the portion of the shuffle operation by transferring data over the P2P connection. The accelerator may be integral with the storage device. The storage node may include a server. The storage device may be a first storage device, the accelerator may be a first accelerator, the P2P connection may be a first P2P connection, and the storage node may further include a second storage device, a second accelerator, and a second P2P connection between the second storage device and the second accelerator, wherein the second storage device and the second accelerator may be configured to perform the portion of the shuffle operation by transferring data over the second P2P connection. The first and second storage devices may be configured to perform the portion of the shuffle operation by transferring data through a direct memory access operation. The storage node may be configured to perform the portion of the shuffle operation by transferring data to an additional storage node through a remote direct memory access operation.

A method for partitioning data may include sampling, at a device, data from one or more partitions based on a number of samples, transferring the sampled data from the device to a host, determining, at the host, one or more splitters based on the sampled data, communicating the one or more splitters from the host to the device, and partitioning, at the device, data for the one or more partitions based on the one or more splitters. The method may further include determining, at the device, a number of records for the one or more partitions, communicating the number of records for the one or more partitions from the device to the host, determining, at the host, the number of samples for the one or more partitions, and communicating the number of samples from the host to the device. The sampling may be performed by an accelerator at the device.

A system may include a storage node comprising an accelerator, and a host configured to perform a shuffle operation for data stored at the storage node, wherein the shuffle operation may include a shuffle write operation and a shuffle read operation, and the storage node may be configured to perform at least a portion of the shuffle operation using the accelerator.

DETAILED DESCRIPTION

Overview

Some of the inventive principles of this disclosure relate to offloading one or more portions of a shuffle operation from a host to a storage node. For example, one or more portions of a shuffle write and/or shuffle read operation may be performed by an accelerator at a storage node. Depending on the implementation details, this may reduce a processing workload at the host and/or reduce input and/or output (I/O) operations and/or network transfers between the host and one or more components of one or more storage nodes.

Some additional inventive principles of this disclosure relate to the use of a peer-to-peer (P2P) connection between a storage device and an accelerator for one or more portions of a shuffle operation. For example, a P2P connection may transfer data between a storage device and an accelerator for one or more portions of a shuffle write and/or shuffle read operation. Depending on the implementation details, this may reduce I/O operations and/or network transfers between the host and one or more components of one or more storage nodes.

Some additional inventive principles of this disclosure relate to the use of one or more types of direct memory access (DMA) operations for one or more portions of a shuffle operation. For example, a DMA or remote DMA (RDMA) operation may be used to transfer data between storage devices within a node or between different nodes. Depending on the implementation details, this may reduce I/O operations and/or network transfers between nodes and/or between storage devices within a node.

The principles disclosed herein may have independent utility and may be embodied individually, and not every embodiment may utilize every principle. Moreover, the principles may also be embodied in various combinations, some of which may amplify the benefits of the individual principles in a synergistic manner.

Shuffle Operations

In some embodiments, a shuffle operation may be used to rearrange data between partitions, devices, and/or nodes in a distributed data processing framework. This may be helpful, for example, when a transformation involves data from other partitions such as summing values in a column.

FIG.1illustrates an embodiment of a shuffle operation for a distributed data processing framework. The shuffle operation102may include at least two parts: a map task and a reduce task. One or more map tasks may receive chunks of collocated data104A,104B, and104C (collectively104), for example, from one or more previous operators such as a map operator, a filter operator, and/or the like. The map task may partition the received data into different blocks106A,106B, and106C (collectively106) based on one or more partitioning rules such as range partitioning, list partitioning, and/or the like. In the embodiment illustrated inFIG.1, a partitioning rule may be implemented as range partitioning108. Thus, the inputs, which in this embodiment may be numerical values, may be assigned to different partitions that may be indicated by different types of shading in portions of the blocks106. The map task may then concatenate the different blocks106into a file and write (e.g., persist) the file into local storage as an intermediate map output.

In a reduce part of the shuffle operation,102one or more reduce tasks may request (e.g., read) blocks from the intermediate map output to constitute a specific partition. Thus, entries from different input chunks that have been assigned to the same partition may be gathered into the same reduce output block110A,110B, or110C (collectively110), which may now be coextensive with partitions, as shown by the different types of shading inFIG.1. After the shuffle operation102, data may have been exchanged such that one or more post-shuffle operations (e.g., sort) may continue with further execution on one or more of the reduce-side partitions110. In some embodiments, the shuffle operation102may ensure the inter-partition ordering and/or indirectly ensure or guarantee the global ordering of the dataset which may include blocks112A,112B, and112C (collectively112), for example, after one or more sort operations114.

Although the embodiment illustrated inFIG.1may be illustrated with three each of the input chunks104, map output blocks106, and reduce output blocks (e.g., partitions)110, any number of blocks may be used at any stage of the shuffle operation102. For example, in some embodiments, if a range partition rule specifies 100 ranges, there may be 100 reduce output blocks110.

In some embodiments, the shuffle operation102may be implemented with at least a shuffle write operation and a shuffle read operation. A shuffle write operation may be performed, for example, by a map task which may rearrange input data into one or more blocks that may include entries belonging to different partitions. The shuffle write operation may write these blocks to local storage as the intermediate map output. A shuffle read operation may be performed, for example, by a reduce task which may obtain a map status that has been logged to a driver by the shuffle write operation. The map status may indicate which blocks of the intermediate map output may contain data entries for each partition. The reduce task may fetch one or more blocks of the intermediate output, or portions thereof, based on the map status. If any of the blocks are located at a different node, the reduce task may fetch those blocks, or portions thereof, across a network.

FIG.2illustrates an embodiment of a shuffle architecture for a distributed data processing framework. In the embodiment illustrated inFIG.2, a map task202(which may also be referred to as a mapper) is illustrated generally on the left side of the figure, and a reduce task204(which may also be referred to as a reducer) is illustrated generally on the right side of the figure and conceptually separated by the shaded bar down the middle of the figure. In some embodiments, essentially all of the processing operations (e.g., computations) may be performed by a central processing unit (CPU).

A shuffle operation may begin when an action in an execute method206in the reduce task204triggers an initiation operation208in the map task202as shown by arrow207. The initiation operation208may initiate a shuffle write operation210with input data and/or shuffle dependency information.

The shuffle write operation210may include a data partition operation212in which the CPU may fetch data from one or more storage devices using one or more I/O operations across a Peripheral Component Interconnect Express (PCIe) interconnect. The data partition operation212may then partition the input data into one or more blocks by assigning a partition identification (ID) to each entry of the input data according to a partition rule.

A data merge operation214may merge data entries belonging to the same partition (e.g., data having the same partition ID) into continuous chunks of map output data. The data merge operation214may also sort and/or aggregate the data, for example, depending on one or more shuffle requirements.

When the amount of processed data reaches a spill threshold, the shuffle write operation210may initiate a data spill sequence. In some embodiments, the data spill sequence may include a data serialize and/or data compression operation216which may reduce the amount of map output data that is transferred through an I/O operation. Then, in a data spill operation218, the CPU may write the map output data to one or more files in local storage using one or more I/O operations. At operation220, the CPU may register map status data, which may include metadata for the map output data, with a driver for the distributed data processing framework. The driver may publish the map status data for use throughout the framework.

The execute method206in the reduce task204may also initiate a shuffle read operation222in which the CPU may request the map status from the driver at operation224. During a fetch operation226, the CPU may then use the map status to fetch one or more blocks, for example, for each partition. If the CPU and input data are located at different nodes, the CPU may fetch the data through a network and/or network protocol such as Ethernet and/or Transmission Control Protocol/Internet Protocol (TCP/IP). In some embodiments, the shuffle read operation222may include a data decompression and/or data deserialize operation228in which the CPU may transform the received data to its original form.

During a data merge operation230, the CPU may merge data entries belonging to the same partition into continuous chunks of reduce output data which the CPU may then write to local storage through one or more I/O operations. In some embodiments, the data merge operation230may also sort and/or aggregate the data, for example, depending on one or more shuffle requirements. The reduce task204may then proceed with one or more post-shuffle operations232such as a sort operation.

As illustrated inFIGS.1and2, in some embodiments, a shuffle operation may involve one or more CPU-intensive operations such as serialize/deserialize, compress/decompress, sorting and/or merging operations. A shuffle operation may also involve heavy I/O operations and/or network transfers of many small fragmented files. Moreover, a partition operation such as operation212may involve additional CPU-intensive sampling operations and/or additional fragmented I/O operations, for example, when data is too large to fit in memory. Additionally, when data is written to and/or read from a storage device, and/or transferred across a network during a shuffle operation, it may halt the ability of a distributed data processing framework to perform other processing. Thus, a shuffle operation may cause a performance bottleneck, especially in large-data, shuffle-intensive applications. Furthermore, shuffle operations may cause stress on a CPU, memory, storage devices, and/or network capacities of one or more devices and/or clusters running a distributed data processing framework.

In some embodiments, it may be beneficial to reduce the number of shuffle operations that are performed or reduce the amount of data that is transferred during a shuffle operation. However, shuffling data in a many-to-many fashion across a network may be non-trivial. In some embodiments, all or most of an entire working set, which may be a large fraction of the input data, may be transferred across the network. This may place a significant burden on an operating system (OS) at the source and/or the destination, for example, by requiring many file and/or network I/O operations.

Shuffle Acceleration

In some embodiments according to this disclosure, one or more portions of a shuffle operation may be offloaded to an accelerator at a storage node.

FIG.3illustrates an embodiment of a system for implementing shuffle acceleration according to this disclosure. The embodiment illustrated inFIG.3may include a CPU302, a solid state drive (SSD)304, a field programmable gate array (FPGA)306, a dynamic random access memory (DRAM)308, a PCIe switch310, and PCIe links312,314, and316. The PCIe topology illustrated inFIG.3may be configured so the SSD304and the FPGA306appear as endpoints to the host CPU302. Thus, the CPU302may be responsible for transferring data between the SSD304and the FPGA306, as well as the DRAM308for the FPGA306.

If the system illustrated inFIG.3is used to implement a shuffle operation such as the one illustrated inFIG.2, one or more processing portions of the shuffle write operation210and/or the shuffle read operation222may be offloaded to the FPGA306. For example, one or more portions of the partition operation212, merge operations214and230, serialize/deserialize and/or compress/decompress operations216and228, data spill operation218and/or fetch operation226may be offloaded to the FPGA306. Depending on the implementation details, offloading processing for one or more portions of these operations may reduce clock cycles consumed by the CPU302. However, offloading operations to the FPGA306may also involve transferring data to and from the FPGA306and DRAM308through the CPU302. This may increase the cost of communication between the components which may impose a latency and/or throughput limitation on the system. Moreover, if multiple SSDs and/or FPGAs are added to the system, the CPU302may lack the ability to scale the communications with the SSDs and/or FPGAs due to high CPU overhead and/or limited PCIe bandwidth.

Some embodiments according to this disclosure may include a P2P connection, which may be implemented as a private connection, between one or more storage devices and one or more accelerators.

FIG.4illustrates an embodiment of a shuffle acceleration system having a P2P connection according to this disclosure. The embodiment illustrated inFIG.4may include a host402, a storage node404, and a communication interface406. The storage node404may include a storage device408, an accelerator410, and a P2P connection412.

The use of a P2P connection such as that illustrated inFIG.4may enable an accelerator to directly access data in one or more storage devices, thereby conserving limited bandwidth on the connection between the CPU and a storage device, and the connection between the CPU and an accelerator. Depending on the implementations details, a P2P connection may increase the bandwidth and/or reduce overhead, memory usage, and/or power consumption associated with transferring data between a storage device and an accelerator compared to transferring data through the host and/or host memory. In some embodiments, a P2P connection may be especially helpful for shuffle acceleration operations which may involve migrating data between an accelerator and a storage device multiple times.

In some embodiments, and depending on the implementation details, implementing computations at, or close to, a storage device (e.g., through the use of an accelerator) may reduce the cost and/or power of I/O operations. It may also increase system scalability, for example, in the context of managing larger storage systems. However, scalability in larger storage systems with multiple storage devices such as SSDs may be limited, for example, by the capacity of host memory and/or CPU overhead involved with reading and/or writing data and/or sharing connection bandwidth. In some embodiments, and depending on the implementation details, a shuffle acceleration technique using P2P communications as disclosed herein may increase the system scalability by removing or mitigating one or more of these bottlenecks.

Referring again toFIG.4, the communication interface406may provide a first communication connection between the host402and the storage device408, and a second communication connection between the host402and the accelerator410. In some embodiments, the first and second communication connections may be implemented through separate physical and logical connections. In some embodiments, the first and second communication connections may be implemented through physical and/or logical connections that may be partially or entirely combined. For example, the storage node404may include a physical switch that may implement the first and second communication connections as partially separate physical connections. As another example, the first and second communication connections may be implemented as separate logical or virtual connections on a combined physical connection.

The communication interface406may be implemented with any type of communication structure and/or protocol. For example, the communication interface406may be implemented entirely or partially with an interconnect structure and/or protocol such as PCIe, Compute Express Link (CXL), Cache Coherent Interconnect for Accelerators (CCIX), and/or the like. As another example, the communication interface406may be implemented entirely or partially with a network structure and/or protocol such as Ethernet, TCP/IP, Fibre Channel, InfiniBand, and/or the like. As a further example, the communication interface406may be implemented entirely or partially with a storage interface and/or protocol such as Serial ATA (SATA), Serial Attached SCSI (SAS), Non-Volatile Memory Express (NVMe), and/or the like. Moreover, any of these structures, protocols, and/or interfaces may be combined in hybrid combinations such as NVMe over fabric (NVMe-oF).

The P2P connection412may be implemented with any type of communication structure and/or protocol such as the interconnect, network, and/or storage interfaces described above. In some embodiments, the P2P connection412may be implemented entirely or partially as a separate logical or virtual connection on a shared physical connection that may be used to implement the communication interface406.

The host402may be implemented with any type of processing apparatus. Examples may include one or more general or special purpose CPUs including complex instruction set computer (CISC) and/or reduced instruction set computer (RISC) processors, and/or the like, as well as FPGAs, application specific integrated circuits (ASICs), systems on chip (SOCs), and/or any other components that may perform the functions of a host processor for a distributed data processing framework such as Apache Spark, Apache Hadoop, and/or the like.

The storage device408may be implemented with any type of storage device such as a hard disk drive (HDD), an SSD, persistent storage such as cross-gridded memory with bulk resistance change, and/or the like.

The accelerator410may be implemented with any type of processing apparatus including one or more CISC and/or RISC processors, FPGAs, ASICs, SOCs, and/or graphics processing units (GPUs), as well as any combinational logic, sequential logic, timers, counters, registers, gate arrays, complex programmable logic devices (CPLDs), state machines, and/or the like. In some embodiments, the accelerator may be implemented as part of a storage controller for the storage device408. In some embodiments, one or more memories such as DRAMs may be provided for, or integral with, the accelerator410to provide workspace memory for one or more portions of a shuffle operation that may be offloaded to the accelerator410.

In some embodiments, the accelerator410may implement some or all of the offloaded shuffle operations primarily in software, for example, running on a general or special purpose CPU. In some embodiments, the accelerator410may implement some or all of the offloaded shuffle operations primarily in hardware. For example, one or more offloaded shuffle operations may be implemented in dedicated logic on an ASIC. As another example, one or more offloaded shuffle operations may be programmed into an FPGA. Depending on the implementation details, implementing offloaded shuffle operations in hardware may provide increased throughput, reduced latency, reduced memory usage, and/or reduced power consumption.

Although shown as a separate component, the host402may be implemented integral with the node404. Similarly, although shown integral with the node404, the storage device408and/or accelerator410may be implemented separate from the node404. In some embodiments, the accelerator410may be integral with the storage device408.

The embodiment illustrated inFIG.4may be implemented in any physical configuration. For example, in some embodiments, the system may be implemented as a server chassis in which the host402, storage device408, and accelerator410are implemented as separate components, and the communication interface406and P2P connection412may be implemented via PCIe links through a backplane, midplane, and/or the like. In such a configuration, the first communication connection between the host402and the storage device408may be implemented as a first point-to-point PCIe link, the second communication connection between the host402and the accelerator410may be implemented as a second point-to-point PCIe link, and the P2P connection412may be implemented as a third point-to-point PCIe link. Alternatively, one or more of the point-to-point PCIe links may be combined through one or more PCIe switches.

In another example physical configuration, the storage node404may be implemented as a server chassis containing the storage device408and the accelerator410, while the host402may be implemented in a separate chassis or rack, or in a remote location. In this configuration, the communication interface406may be implemented with a network structure and/or protocol such as Ethernet and TCP/IP, and the storage device408may be implemented as an Ethernet SSD (eSSD). Additionally, in this configuration, a network switch may be provided on a backplane, midplane, switchboard, and/or the like, to provide connectivity between the storage device408and the host402and/or between the accelerator410and the host402. In this configuration, the P2P connection412may be implemented, for example, through a point-to-point PCIe, or through a PCIe switch on a backplane, midplane, switchboard, and/or the like. Alternatively, or additionally, the P2P connection may be implemented as a logical connection through a network switch as described above.

FIG.5illustrates an example embodiment of a shuffle acceleration system having a P2P connection according to this disclosure. The embodiment illustrated inFIG.5may include a host502, a storage node504, a communication interface506a storage device508, an accelerator510, and a P2P connection512which may be similar to those illustrated inFIG.4and which may be implemented using any of the implementation details disclosed above. However, the embodiment illustrated inFIG.5may include a switch514which may implement a first communication connection516between the host502and the storage device508, and a second communication connection518between the host502and the accelerator510. In some embodiments, the switch514may be implemented a network switch such as an Ethernet switch, an interconnect switch such as a PCIe switch, and/or the like.

FIG.6illustrates another example embodiment of a shuffle acceleration system having a logical P2P connection through a switch according to this disclosure. The embodiment illustrated inFIG.6may include a host602, a storage node604, a communication interface606a storage device608, an accelerator610, and a switch614which may be similar to those illustrated inFIGS.4and5and which may be implemented using any of the implementation details disclosed above. However, in the embodiment illustrated inFIG.6, the P2P connection612between the storage device608, and the accelerator610may be implemented as a logical connection through the switch614.

FIG.7illustrates another example embodiment of a shuffle acceleration system having a logical P2P connection through a switch that is integral with an accelerator according to this disclosure. For purposes of illustration, the embodiment shown inFIG.7may include some specific implementation details such as a PCIe interconnect topology, an FPGA accelerator, and an SSD storage device. In other embodiments, however, these components may be replaced with substitutes such as HDDs, storage drives based on persistent cross-gridded memory with bulk resistance change, and/or the like, interconnect or network interfaces such as CXL, Ethernet, and/or the like, and ASICs, SOCs, and/or the like.

In the embodiment illustrated inFIG.7, a storage node may be implemented as an enhanced SSD702having an SSD controller704, a not-AND (NAND) flash memory medium706, an FPGA708, and a DRAM710. The FPGA708may include a PCIe switch712and an accelerator714. The PCIe switch712may be physically connected to a CPU (Host)716, the SSD controller704, and the accelerator714through PCIe links718,720, and722, respectively. However, the switch712may implement logical or virtual connections724,726, and728. Logical connection724may handle data read and/or write transfers between the SSD controller704, which may appear as a separate endpoint, and the host716. Logical connection726may handle data read and/or write transfers between the accelerator714, which may appear as a separate endpoint, and the host716. Logical connection728may handle P2P communications between the SSD controller704and the accelerator714. In some embodiments, one or more switch functions may be implemented as soft switch functions.

The PCIe switch712may be physically integrated into the FPGA708for convenience and/or availability of integrated circuit (IC) area. In other embodiments, however, the PCIe switch712may be a separate component or may be integrated into the SSD controller704. In other embodiments, any number of the components illustrated inFIG.7may be integrated on an SOC.

FIG.8illustrates a PCIe base address register (BAR) map that may be used to implement the switching functions of the PCIe switch712. In the address map802, an NVMe register address range804may support NVMe transfers between the SSD controller704and the host716. An accelerator address range806and a DRAM address range808may support data transfers between the host716and the accelerator714and DRAM710. In some embodiments, the DRAM for the FPGA may be exposed to the host PCIe address space. In some embodiments, NVMe commands may securely stream data between the SSD controller704and the FPGA708using the P2P connection728.

The embodiment illustrated inFIG.7, which may be used to implement any of the shuffle acceleration techniques disclosed herein, may provide enhanced, and in some implementations, unlimited concurrency. In some embodiments, and depending on the implementation details, this configuration may: conserve cache (e.g., L2:DRAM) bandwidth; enable scaling without expensive CPUs at storage nodes; and/or avoid funneling and/or data movement to and/or from standalone accelerators.

FIG.9illustrates an embodiment of a shuffle architecture having shuffle acceleration according to this disclosure. The embodiment illustrated inFIG.9may include a map task902including a shuffle write operation910and a reduce task904including a shuffle read operation922. These tasks and shuffle operations may include some elements similar to those illustrated inFIG.2which may have similar numbers and may operate in a similar manner. However, in the system illustrated inFIG.9, one or more portions of the shuffle operation, for example, CPU-intensive operations, may be offloaded to an accelerator. In some embodiments, all or part of any of the operations shown with shading may be offloaded to an accelerator, while operations shown without shading may be executed by a CPU. For example, in the shuffle write operation910, all or part of a data partition operation912, a data merge operation (which may include a data sort and/or a data aggregation operation)914, a data serialize and/or data compression operation916, and/or a data spill operation918may be offloaded to an accelerator. As another example, in the shuffle read operation922, all or part of a fetch operation926, data decompression and/or data deserialize operation928and/or data merge operation (which may include data a sort and/or a data aggregation operation)930may be offloaded to an accelerator. Examples of a serialize operation may include converting an object or other data to a bitstream or byte stream for transfer to a storage device. Examples of a deserialize operation may include converting a bitstream or byte stream back to an object or other data after being transferred from a storage device. Serialize and/or deserialize operations may be performed, for example, using serialize/deserialize features in Java, Kryo, and/or the like. In the embodiment illustrated inFIG.9, and any other embodiments disclosed herein, the labeling of an element as optional does not indicate that other elements are mandatory.

The embodiment illustrated inFIG.9may be implemented with any acceleration architecture according to this disclosure including those illustrated inFIGS.3-8. If implemented with a system having P2P communication between a storage device and the accelerator, any or all of the partition operation912, the data spill operation918and/or the data merge operation930may utilize the P2P connection. For example, the data merge operation918may write intermediate map output data directly to local storage such as an SSD through the P2P connection rather than using relatively expensive I/O operations with a CPU. As another example, the partition operation912may use the P2P connection to write data to local storage as described below. Depending on the implementation details, the use of the P2P connection may increase throughput and/or reduce cost, latency, memory usage, and/or power consumption compared to using relatively expensive I/O operations with a CPU.

The embodiment illustrated inFIG.9may also use one or more direct memory access (DMA) techniques. For example, the fetch operation926may fetch one or more blocks of map output data based on node, partition, and/or block information using DMA if the data is located within the same node (e.g., in the same storage device, or in a storage device in the same server chassis), or using remote DMA (RDMA) across a network if the data is located at a different node. In some embodiments, and depending on the implementation details, the use of DMA and/or RDMA may increase throughput and/or reduce cost, latency, memory usage, and/or power consumption compared to using a protocol such as TCP/IP across a network and/or I/O operations with a CPU.

Referring again toFIG.9, once the data blocks are fetched using DMA/RDMA and stored in local storage, the data merge operation930may use the private P2P connection to transfer data from local storage (e.g., from an SSD to FPGA DRAM) to merge data and continue executing one or more post-shuffle operations932, which, in some embodiments, may also be offloaded to an accelerator.

FIG.10illustrates an embodiment of a system having a P2P connection and a DMA/RDMA engine according to this disclosure. The embodiment illustrated inFIG.10may include a host1002, a storage node1004, a communication interface1006, a storage device1008, an accelerator1010, and a P2P connection1012which may be similar to those illustrated inFIG.4. The embodiment illustrated inFIG.10, however, may also include a DMA/RDMA engine1014which may enable the storage node1004to perform DMA transfers, for example, using an interconnect such as PCIe with devices located at the same node. The DMA/RDMA engine may also enable the storage node1004to perform RDMA transfers, for example, using a network such as Ethernet with devices located at different nodes.

FIG.11illustrates an embodiment of a method of processing data in a system having a host and a storage node according to this disclosure. The method may begin at operation1102. At operation1104, the method may perform a shuffle operation on data stored at the storage node. Operation1104may include a suboperation1106, which may include performing a shuffle write operation. Operation1104also may include a suboperation1108, which may include performing a shuffle read operation. In operation1104, at least a portion of the shuffle operation is performed by an accelerator at the storage node. The method may terminate at operation1110.

The operations and/or components described with respect to the embodiment illustrated inFIG.11, as well as all of the other embodiments described herein, are example operations and/or components. In some embodiments, some operations and/or components may be omitted and/or other operations and/or components may be included. Moreover, in some embodiments, the temporal and/or spatial order of the operations and/or components may be varied.

Partitioning Operations

FIG.12illustrates an embodiment of a partition operation for a shuffle write operation. In some embodiments, the operation illustrated inFIG.12may be performed by a CPU. At operation1202, the CPU may read each partition and generate data samples for each partition. At operation1204, the CPU may sort the samples and generate splitters based on the samples. In some cases, partitions may be resampled to generate the splitters. At operation1206, the CPU may read each partition again and partition the data based on the splitters. At operation1208, the CPU may continue with other operations in the shuffle write operation.

In some embodiments described above, some offloaded shuffle operations, or portions thereof, may execute concurrently on multiple storage nodes. However, in some embodiments, partitioning may share information which may prevent concurrent execution.

An example embodiment of a range-based partitioning algorithm may proceed as follows: (1) All or some of a dataset may be sampled to obtain K*N samples, where K may be an oversampling factor, which may be any constant value, and N may be the total number of partitions generated after partitioning. (2) An array of K*N samples may be sorted in ascending order. (3) N−1 splitters may be obtained from the sorted K*N samples, for example, by picking a number at every K elements in the array. (4) All or some of the dataset may be partitioned, for example, by directly iterating through the splitters (if N is small), using a binary-search-tree (e.g., if N is large) in a record-by-record fashion, and/or the like.

In some embodiments, a partitioning algorithm may generate evenly-sized partitions for the reducer, thus the sampled data may well represent the entire dataset distribution, which may mean, for example, that the more records reside in a map side partition, the more samples may be generated from that partition. In some applications (e.g., big data applications), data may be distributed among multiple nodes and multiple storage devices. To offload partitioning to an accelerator device while avoiding the overhead to transfer a large amount of data, an embodiment of a workflow design according to this disclosure may reduce or minimize CPU work for coordination.

FIG.13illustrates an example embodiment of a partitioning workflow architecture according to this disclosure. In the embodiment illustrated inFIG.13, operations performed by a host (e.g., a CPU) are illustrated generally on the left side of the figure, and operations performed at a storage node (which may include an accelerator and/or a P2P connection between the accelerator and a storage device) are shaded and illustrated generally on the right side of the figure and conceptually separated by the shaded bar down the middle of the figure.

At operation1302, the storage node may determine a number of records for the one or more partitions. The storage node may communicate the number of records to the host at communication 1. At operation1304, the host may determine the number of samples that should be collected on a per partition basis for one or more of the partitions. The host may communicate the number of samples per partition to the storage node at communication 2. At operation1306, the storage node may sample data from one or more partitions based on the number of samples determined by the host. The storage node may transfer the sampled data to the host at communication 3. At operation1308, the host may sort the sampled data and determine one or more splitters based on the sampled data. The host may communicate a set of one or more splitters to the storage node at communication 4. At operation1310, the storage node may partition the data locally into one or more partitions based on the set of splitters. At operation1312, the storage node may continue with other steps in a shuffle write operation.

In some embodiments, and depending on the implementation details, the principles illustrated inFIG.13may provide any number of the following advantages. First, some CPU-intensive operations such as scanning, binary search, sampling a large dataset, and/or the like, may be offloaded to one or more accelerators, which may lower the CPU utilization. Second, data transfer between a host and a storage node may be reduced or minimized, for example, by communicating relatively small amounts of information such as the size of each partition, the number of samples, and/or the sample data themselves, which may be smaller than the entire dataset. Additionally, in some embodiments, an overall architecture according toFIG.13may reduce memory consumption and/or CPU utilization, for example, by efficiently offloading CPU-intensive tasks. Data parallelism in an accelerator may also be properly preserved with reduced or minimum CPU coordination and/or scheduling.

Referring again toFIG.13, in some embodiments, single nodes (e.g., storage devices) may have multiple partitions. Multiple partitions in a single storage device may be iterated through in a round-robin fashion, for example, controlled by a single CPU thread. In some embodiments, traffic loads (e.g., in communications 1-4) may be smaller compared to scanning all records in all partitions. In some embodiments, one or more nodes may know the number of records for each partition on it, as well as the total number of records across some or all nodes, which may be used, for example, to determine the number of samples to generate for each partition on it proportionally.

The embodiment illustrated inFIG.13may be described in the context of a shuffle operation using an accelerator, but the inventive principles may be applied in other applications. For example, the principles illustrated inFIG.13may be applied to a sort operation such as an external sort, which may involve sorting a large amount of data that may be stored in one or more storage devices. In such an embodiment, a range partition rule may employ P2P connections between storage devices and accelerators to avoid transferring large amounts of data back-and-forth between the storage devices and a host. In some embodiments, and depending on the implementation details, this may improve parallelism, performance, and/or energy efficiency, and may reduce CPU and/or memory usage and/or network overhead.

As another example, the principles illustrated inFIG.13may be applied to range partitioning for a database. For example, one or more tables, which may be the input of one or more database operations, may be partitioned by range, which may be used in a database engine to improve or optimize data storage and/or query performance.

In some embodiments, the principles of this disclosure may provide a generic architecture for shuffle acceleration. Some embodiments may use one or more accelerators (e.g., storage device near-storage computing power) and/or P2P data transfer via a private interconnect between an accelerator device and storage device, as well as utilizing DMA and/or RDMA engines in some implementations to reduce I/O and/or CPU costs. In some embodiments, and depending on the implementation details, a near-storage-accelerated shuffle architecture may provide any number of the following features and/or benefits.

An architecture according to this disclosure may use an enhanced storage device having computational capabilities and/or an accelerator device to accelerate a shuffle operation, which may improve the performance of data-intensive and/or shuffle-intensive applications. Some embodiments may reduce I/O costs, memory consumption, CPU utilization, network overhead and/or the like.

P2P communication between a storage device and accelerator device via private interconnect may improve the scalability of a system, for example, by not overwhelming limited interconnect bandwidth to a host CPU.

Some embodiments may be implemented as a generic shuffle acceleration architecture. As a shuffle operation may be a necessity in some systems, and a bottleneck, for example, in data processing platforms (e.g., big data), some embodiments may have broad prospects in many applications.

In some embodiments, an accelerator device implementation such as an FPGA or application specific integrated circuit (ASIC) may have less power consumption, for example, as compared to a general-purpose processor, which may increase the overall energy efficiency.

The embodiments disclosed above have been described in the context of various implementation details, but the principles of this disclosure are not limited to these or any other specific details. For example, some functionality has been described as being implemented by certain components, but in other embodiments, the functionality may be distributed between different systems and components in different locations and having various user interfaces. Certain embodiments have been described as having specific processes, steps, etc., but these terms also encompass embodiments in which a specific process, step, etc. may be implemented with multiple processes, steps, etc., or in which multiple processes, steps, etc. may be integrated into a single process, step, etc. A reference to a component or element may refer to only a portion of the component or element. For example, a reference to an integrated circuit may refer to all or only a portion of the integrated circuit, and a reference to a block may refer to the entire block or one or more subblocks. The use of terms such as “first” and “second” in this disclosure and the claims may only be for purposes of distinguishing the things they modify and may not to indicate any spatial or temporal order unless apparent otherwise from context. In some embodiments, based on” may refer to “based at least in part on.” In some embodiments, “disabled” may refer to “disabled at least in part.” A reference to a first thing may not imply the existence of a second thing.

Various organizational aids such as section headings and/or the like may be provided as a convenience, but the subject matter arranged according to these aids and the principles of this disclosure and the embodiments described herein are not defined or limited by these organizational aids.

The various details and embodiments described above may be combined to produce additional embodiments according to the inventive principles of this patent disclosure. Since the inventive principles of this patent disclosure may be modified in arrangement and detail without departing from the inventive concepts, such changes and modifications are considered to fall within the scope of the following claims.