Data caching in a large-scale processing environment

Systems, methods, and software described herein facilitate an enhanced service architecture for large-scale data processing. In one implementation, a method of providing data to a large-scale data processing architecture includes identifying a data request from a container in a plurality of containers executing on a host system, wherein the plurality of containers each run an instance of a large-scale processing framework. The method further provides identifying a storage repository for the data request, and accessing data associated with the data request from the storage repository. The method also includes caching the data in a portion of a cache memory on the host system allocated to the container, wherein the cache memory comprises a plurality of portions each allocated to one of the plurality of containers.

TECHNICAL FIELD

Aspects of the disclosure are related to computing hardware and software technology, and in particular to the management of virtual clusters in a large-scale processing environment.

TECHNICAL BACKGROUND

An increasing number of data-intensive distributed applications are being developed to serve various needs, such as processing very large data sets that generally cannot be handled by a single computer. Instead, clusters of computers are employed to distribute various tasks, such as organizing and accessing the data and performing related operations with respect to the data. Various applications and frameworks have been developed to interact with such large data sets, including Hive, HBase, Hadoop, Spark, Amazon S3, and CloudStore, among others.

At the same time, virtualization techniques have gained popularity and are now commonplace in data centers and other computing environments in which it is useful to increase the efficiency with which computing resources are used. In a virtualized environment, one or more virtual nodes are instantiated on an underlying host computer and share the resources of the underlying computer. Accordingly, rather than implementing a single node per host computing system, multiple nodes may be deployed on a host to more efficiently use the processing resources of the computing system. These virtual nodes may include full operating system virtual machines, Linux containers, such as Docker containers, jails, or other similar types of virtual containment nodes. However, although virtual nodes may more efficiently use the resources of the underlying host computing systems, difficulties arise in scaling the virtual nodes and providing data from large storage repositories to the individual virtual nodes.

Overview

The technology disclosed herein facilitates an enhanced service architecture for large-scale data processing. In one implementation, a method of providing data to a large-scale data processing architecture includes identifying a data request from a container in a plurality of containers executing on a host system, wherein the plurality of containers each run an instance of a large-scale processing framework. The method further provides identifying a storage repository for the data request, and accessing data associated with the data request from the storage repository. The method also includes caching the data in a portion of a cache memory on the host system allocated to the container, wherein the cache memory comprises a plurality of portions each allocated to one of the plurality of containers.

TECHNICAL DISCLOSURE

Large-scale processing environments (LSPEs) may employ containers running via host computing systems to provide efficient execution of job processes. These containers, which may include Linux containers, Docker containers, jails, and like, each contain the necessary large-scale processing framework (LSPF) and dependencies required to perform operations on a target data set. Rather than allocating an operating system to each of the processing nodes within the LSPE, containers may share a host operating system and common resources, such as files, databases, and common processes.

In addition to the containers, the LSPE further includes data sources that may be stored on the same physical computing systems as the containers or on separate physical computing systems and devices that are accessible using the internet or some other similar network. These data sources may be stored using Hadoop distributed file system (HDFS), versions of the Google file system, versions of the Gluster file system (GlusterFS), or any other distributed file system version—including combinations thereof. Data sources may also be stored using object storage systems such as Swift.

To assign job processes, such as Apache Hadoop processes, Apache Spark processes, Disco processes, or other similar job processes to the host computing systems within a LSPE, a control node may be maintained that can distribute jobs within the environment. The control node may identify a job and preferences associated with the job, such as an amount of processing resources, number of nodes, and the like, and allocate containers as required by the particular job process. Once allocated, the container nodes may execute and provide the desired operations on a data set.

In the present example, to provide the desired operations on the data stored within the data sources, a cache service is provided capable of acting as an intermediary between the containers and the data sources. This cache service, which is shared between the containers, identifies a data request from a container, determines a data source associated with the request, accesses the data associated with the request from the data source, and caches the data within allocated cache memory accessible by the requesting container. This cache memory allows the containers to directly access data from the host without first caching the data to the host and transferring the data to a location associated with the container. In some implementations, the cache memory may be allocated a portion of physical memory on the host computing system, which can be distributed as sub-portions to the individual containers. This allocation of physical memory includes addresses accessible by the containers to retrieve and write data for LSPF operations.

Referring now toFIG. 1,FIG. 1illustrates a computing architecture100to cache data for large-scale processing containers. Computing architecture100includes host system105and data sources140. Host system105further includes operating system110, cache service115, and containers120-122. Data sources140further includes data repositories141-142. Although illustrated separate in the present example, it should be understood that data repositories141-142might be stored wholly or partially on host system105. However, when stored separately, host system105may communicate with data repositories141-142using the internet or some other communication network.

In operation, containers120-122are initiated on host system105to provide large scale processing operations on job processes allocated to the computing environment. These job processes may include Apache Hadoop job processes, Apache Spark job processes, or any other similar job processes. To provide the operations for the job processes, containers120-122include large-scale processing (LSP) operations130-132, which correspond to the necessary LSPFs to provide the operations desired by the cluster. In addition to the containers, each of the containers share cache service115, which comprises a utility to cache data for operations within the containers. In particular, containers120-122are capable of sharing operating system110, and may further share one or more repositories and/or applications from the host system. Accordingly, rather than a separate cache service for each of the containers, cache service115may include one or more repositories that can be shared by the containers.

In the present example of cache service115, cache service115includes the ability to cache data to cache memory addresses for containers120-122. This cache memory permits data to be cached for each of the containers within shared memory addresses on host system105. In some implementations, cache service115may identify a request from an LSP operation in one of the containers. In response to the request, cache service115identifies the appropriate data repository associated with the data request, and accesses the necessary data from the data repository. Once accessed, the required data is placed in cache memory locations by cache service115, permitting the requesting container to access the requested data.

In some implementations, cache service115may be provided with a portion of physical memory from host system105that can be allocated to containers120-122when a job process is assigned to the containers. This physical memory may include dynamic random-access memory (DRAM), a solid-state memory cache on host system105, or some other memory cache for data for the containers. Using the example above, when data is retrieved for the requesting container, the data may be cached in address locations of physical memory associated with the cache service and the specific requesting container. Once cached, the requesting container may process the data from the cache memory and process the data as required.

Referring now toFIG. 2,FIG. 2illustrates a method200of providing data to a large-scale data processing architecture according to one implementation. Method200is demonstrated in the example below using elements and systems of computing architecture100fromFIG. 1.

As illustrated inFIG. 2, method200includes executing containers configured to provide LSP operations (201). These operations may include processes related to Apache Hadoop, Apache Spark, or some other large-scale processing operation. The large-scale processing operations are used by varying job processes to efficiently process and manipulate large data sets.

In addition to executing the containers, method200further includes executing a cache service shared between the containers configured to retrieve and cache data requested by the containers in a portion of memory allocated to cache data for the containers (202). As depicted inFIG. 1, cache service115may comprise a shared resource accessible by each of the containers executing on host system105. This shared resource may include one or more binaries and libraries that assist the containers in retrieving and writing data to repositories for the job process. In particular, referring still toFIG. 1, a LSP operation within a container may generate a data request for a segment of data. As a request is generated by the LSP operation, cache service115identifies the request and identifies a corresponding repository of repositories141-142for the request. Once a repository is identified, cache service115may access the data and cache the data in cache memory locations accessible and allocated to the requesting container.

In at least one implementation, cache service115may be provided with memory locations that can be allocated to the individual containers and LSP operations. For example, when a job process is allocated to containers120-122on host system105, cache service115may allocate a portion of the cache memory to each of the containers, permitting the containers to use cache memory for data processing.

As an illustrative example, container120may request data for a job process using LSP operation130. As the request is generated by container120, cache service115identifies the data request, and determines the appropriate data repository to support the data request. In some implementations, data may be stored in multiple repositories with multiple file system or data object formats. These formats may include, but are not limited to, versions of the Hadoop distributed file system (HDFS), versions of the Google file system, versions of the Gluster file system (GlusterFS), or versions of Swift data storage. To determine the appropriate data repository to support the request, cache service115identifies the request using a first data access format. This data access format may comprise a format associated with the LSPF of container120, such as HDFS for a Hadoop LSP operation, or may comprise an arbitrary data access format that can be used across varying LSPFs. Once the request is received in the first format, cache service115identifies a repository to accommodate the request and translates the first access format into the format required by the identified repository. Thus, if data repository142were required, the first access format may be translated into an access format for data repository142. Once translated, cache service115may access the data and cache the data within the portion of cache memory associated with container120, making the data accessible to requesting container120.

In some implementations, each container of containers120-122may be allocated a quality of service. This quality of service may be based on the job process for the container, may be based on the tenant or user that generated the job process for the container, or may be based on any other similar quality of service. Based on the quality of service associated with each of the containers or the job process, cache memory may be allocated to the containers to cache data for processing during the job process. Accordingly, cache service115may be provided with a portion of physical memory, which in response to an assignment of a job process may be allocated to the individual containers based on the defined quality of service. For example, if container120were allocated a higher quality of service than containers121-122, container120may be provided with a larger share of cache memory than containers121-122.

Although demonstrated in the previous example with a data read request from the containers, it should be understood that cache service115might write data to repositories141-142in data sources140. This write process may include cache service115identifying new data within the cache memory to be written to a data repository, identifying the appropriate repository to store the data, and accessing the identified repository to store the data.

Turning toFIG. 3,FIG. 3illustrates an allocation of physical memory300to large-scale data processing containers and a cache service. Physical memory300is representative of the physical memory for a host system and includes cache service memory330and container memory320-322. Cache service memory330is representative of the physical memory to operate a cache service to retrieve and write data for LSP operations in containers on a host system. Container memory320-322is representative of the physical memory allocated to containers, permitting the containers to execute on the host computing system.

As described herein, containers are initiated on a host computing system to provide large-scale processing operations for an organization. These containers are allocated resources, including memory and processing cycles, which prevent the operations of an individual container from being integrated with the operations of other containers on the physical host computing system. In the present example, in addition to the memory that is allocated to the containers, the containers may be provided with cache memory335that is capable of caching data for the processes of the containers. This data may include data that is to be read by the containers, data that is to be written to storage repositories from the containers, or any other caching of data for the containers.

Here, cache service memory330represents memory of a cache service that is used to identify data requests for the LSP operations within the containers. Once a request is identified for a LSP operation in a container, the cache service determines a storage location in one or more storage repositories to retrieve the required data, and caches the data within cache memory335. For example, if the container associated with container memory320requested a segment of data from a HDFS repository, the cache service would identify the required data segment, retrieve the data from the HDFS repository, and cache the data within a portion of cache memory335allocated to the container. Once the data is cached, the container may access and process the data as required.

In some implementations, the containers may use a file or object request format that is different from the format that is required for the storage repository. Consequently, when the cache service identifies a request in a first format, the cache service may translate the access request to the required format for the data repository, and access the data using the required format. Once the data is accessed, the cache service may cache the data in physical memory300in a manner expected by the container using first access format. Accordingly, if the containers were using an arbitrary access format, the cache service may identify the request in the arbitrary format, identify the format required for the storage repository, translate the request into the format required for the storage repository, and access the data using the required format. Once the data is accessed, the data may be cached in addresses of cache memory335allocated to the particular container.

As depicted inFIG. 3, each of the containers may be allocated a portion of cache memory335, which is separate from the other containers on the same physical host. Further, although not illustrated in the present example, it should be understood that the amount of cache memory335allocated to each of the containers might be different based on a quality of service associated with the container. For example, the cache allocated to the container associated with container memory320might be larger than the cache memory allocated to the containers associated with container memory321-322. This quantity of cache memory for each container may be dictated based on a quality of service for the container, based on the job process allocated to the container, based on the tenant or user that generated the job process, or determined in any other similar manner. In some implementations, cache memory335may be allocated to the individual containers by the cache service when the job processes are initiated on the containers.

Although not illustrated in the present example, it should be understood that in some implementations, the cache service associated with cache service memory330and the containers associated with container memory320-322may execute within a virtual machine of a host computing system. Consequently, when allocating the memory for cache memory335, cache service memory330may allocate physical memory that is associated with the virtual machine to the containers. For example, the cache service may be provided with three gigabytes of memory from the virtual machine to be allocated to the containers, and provide portions of the three gigabytes to each of the containers when they are initiated for a job process.

Turning toFIG. 4,FIG. 4illustrates an operational scenario400of providing data to large-scale processing containers using a cache service. Operational scenario400includes host system410and data sources440. Host system410further includes containers420-421and cache service450. Containers420-421include LSPF resources and job processes435-436. Job processes435-436use LSPF resources430-431to execute data operations in parallel with other processing nodes within the computing environment.

In operation, containers420-421are generated with LSPF resources430-431to provide a platform for multi-node clustered data processing. These containers are provided with separate user space and physical memory on the host system410, permitting each of the nodes to operate separately on the host computing system. In the present example, in addition to sharing an operating system resource provide by host system410, which often includes a Linux operating system, the containers are further configured to share a cache memory space represented in operational scenario400by cache memory453. To provide the required data to cache memory453, cache service450is provided, which comprises processing instructions to identify data requests of the containers and accesses the data from data sources440. Although illustrated as separate from the containers in the present example, it should be understood that cache service450might comprise a process that can be executed at least partially within the containers to identify the data request.

Once the containers are initiated on the host computing system and provided with the job processes, the job processes may require access to data sources to retrieve and store data. In the present implementation, job process435on container420generates a data request to retrieve data from data sources440. Once a data request is generated, cache service450identifies the request, and determines a particular data repository to accommodate the request. In some implementations, the processing clusters may request data from a variety of data repositories, wherein the data repositories may store data in a variety of file system and object formats. These formats may include versions of HDFS, versions of the Google file system, versions of GlusterFS, versions of Swift data storage, or any other similar distributed file system or object based storage. Accordingly, because the nodes may access repositories of varying file system and object storage formats, the cache service may be required to translate each of the requests for the job process into the appropriate access format for the data repository.

For example, if job process435comprised a Swift job process attempting to access data in a HDFS storage repository. Cache service450would identify the request in the Swift access format and translate the request into the access format appropriate for the HDFS storage repository. In the example of operational scenario400, cache service450identifies that data repository441is the appropriate repository to satisfy the data request. This identification may be made using a namespace included in the data request, a file path included in the data request, credentials included in the data request, an address associated with the data request, or by any other similar identification methods. Once identified, cache service450ensures that the access format is appropriate for data repository441and accesses the data from the repository.

After accessing the data, cache service450caches the data in memory addresses accessible to container420and job process435. In particular, cache service450is provided with cache memory that can be allocated to job processes and containers as they are required. For example, when containers420-421and job processes435-436are initiated on host system410, cache service450may allocate portions of cache memory to each of the containers. Once allocated, data for requests by container420may be placed in memory addresses associated with cache memory453to be provided to job process435.

Although illustrated in the present example as retrieving data from the data sources, it should be understood that similar operations apply to writing data to the data sources. For example, if job process435were to write to data repository441, cache service450would identify the write and the associated data within cache memory453. Once the data is identified, cache service450may identify the appropriate access format to write to data repository441. After identification of the appropriate access format, cache service450may execute the write to data repository441.

Turning toFIG. 5,FIG. 5illustrates an operational scenario500of providing data to cache memory in a large-scale data processing environment. Operational scenario500includes physical memory510representative of memory for a physical host computing system, and further includes data repository550representative of one or more computing systems and storage systems capable of storing data for processing within a LSPE. Physical memory510further includes container memory520-522, cache service memory530, and cache memory535.

As described herein, containers are initiated on a host computing system to efficiently use the processing resources provided by the host computing system. To provide data to the LSP operations on the individual application containers, a cache service is provided, which identifies requests from the containers and retrieves the data for the containers. This cache service may comprise a shared function provided to the containers capable of identifying and accessing the appropriate data for the container. In the present example, the LSP operation associated with container memory520requests a segment of data555from data repository550. To accommodate the request, the cache service associated with cache service memory530accesses the data and caches the data in a portion of cache memory535associated with the container. Once the data is cached within cache memory535, the LSP operations within the container may access the data and provide the required operations.

Although illustrated as identical in size in the present implementation, it should be understood that in some examples, the amount of cache memory535allocated to each of the containers may be dynamic based on a variety of factors. Such factors may include, but are not limited to, the size of the job associated with the container, the quality of service associated with the container, or the quality of service associated with the particular job process. Accordingly, if the container corresponding to container memory520were allocated a better quality of service than the containers associated with container memory521-522, then the cache memory535available to the container for container memory520may be larger than for the other containers.

Despite being illustrated in the example of operational scenario500with a read request for the container, it should also be understood that similar operations apply to write requests from the container. Instead of retrieving the required segment of data555, the cache service may identify that a write occurred for the application container, identify the data that was written to cache memory535, and initiate a write process to data555. This write process may include translating the write request into the appropriate format for data repository550in some implementations.

Further, although not illustrated in the present example, it should be understood that physical memory510may comprise memory allocated to a host virtual machine capable of executing the containers associated with container memory520-522and the cache service associated with cache service memory530. In particular, a physical computing system, such as a serving computer, a desktop computer, or some other similar computing system may execute virtual machines to more efficiently use the processing resources of the physical computing system. These virtual machines or hosts for the containers may be provided with a segment of physical memory that can be allocated to the containers, the cache service, and the cache memory for the LSP operations.

FIG. 6illustrates a computing system600for providing data to large-scale data processing containers. Computing system600is representative of any computing system or systems with which the various operational architectures, processes, scenarios, and sequences disclosed herein for a host computing system may be implemented. Computing system600is an example of host system105and host system410, although other examples may exist. Computing system600may comprise one or more server computing systems, desktop computing systems, routers, gateways, switches, and other similar computing elements, including combinations thereof. Computing system600comprises communication interface601, user interface602, and processing system603. Processing system603is linked to communication interface601and user interface602. Processing system603includes processing circuitry605and memory device606that stores operating software607.

Communication interface601comprises components that communicate over communication links, such as network cards, ports, radio frequency (RF) transceivers, processing circuitry and software, or some other communication devices. Communication interface601may be configured to communicate over metallic, wireless, or optical links. Communication interface601may be configured to use Time Division Multiplex (TDM), Internet Protocol (IP), Ethernet, optical networking, wireless protocols, communication signaling, or some other communication format—including combinations thereof. In some implementations, communication interface601is configured to communicate with one or more other computing systems capable of hosting storage repositories for the LSP operations. Further, in some implementations, computing system600may be communicatively coupled to a configuration or management system that configures the containers and processing jobs that execute on computing system600.

User interface602comprises components that interact with a user to receive user inputs and to present media and/or information. User interface602may include a speaker, microphone, buttons, lights, display screen, touch screen, touch pad, scroll wheel, communication port, or some other user input/output apparatus—including combinations thereof. User interface602may be omitted in some examples.

Processing circuitry605comprises microprocessor and other circuitry that retrieves and executes operating software607from memory device606. Memory device606comprises a non-transitory storage medium, such as a disk drive, flash drive, data storage circuitry, or some other memory apparatus. Memory device606comprises cache memory610, which may be allocated to containers609by cache service module608. Processing circuitry605is typically mounted on a circuit board that may also hold memory device606and portions of communication interface601and user interface602. Operating software607comprises computer programs, firmware, or some other form of machine-readable processing instructions. Operating software607includes cache service module608and containers609, although any number of software modules may provide the same operation. Operating software607may further include an operating system, utilities, drivers, network interfaces, applications, or some other type of software. When executed by processing circuitry605, operating software607directs processing system603to operate computing system600as described herein.

In particular, containers609are executed by processing system603to provide LSP cluster operations for one or more tenants of a computing environment. These containers are configured with Hadoop processes, Spark processes, or any other similar process that can distribute a data processing operation across one or more processing nodes. In the present implementation, in addition to the containers, software607further includes cache service module608. Cache service module608directs processing system603to identify data requests for containers609, determine a storage repository associated with the data request, and access the data associated with the request from the determined storage repository. In some implementations, to access the storage repository, cache service module608directs processing system603to translate the access format for the data request from the container, into an access format for the storage repository. Consequently, if the container were to request data in accordance with GlusterFS, but the data were stored in a HDFS repository, cache service module608may direct processing system603to translate the request into a format capable of accessing the HDFS repository.

Once the data is accessed via cache service module608, cache service module608directs processing system603to cache the data in a portion of cache memory610allocated for the requesting container. In particular, when job processes or containers are initiated on the host computing system, cache service module608allocates a portion of available cache memory to each of the containers. In some implementations, these portions of cache memory may be allocated based on a quality of service associated with the job processes or containers on the host. Accordingly, if a job process on one container required a better quality of service than the other containers on the same host, that container may be provided with a larger portion of cache memory than the other containers.

In addition to providing the operations to read data into cache memory610, cache service module608further directs processing system603to initiate write processes to the storage repositories associated with the LSP operations. In particular, when a write is detected by cache service module608, cache service module608determines an appropriate storage repository for the write, makes any necessary translations of the access format, and initiates the write to the storage repository from cache memory610. This may include transferring the data from cache memory610to the storage repository and specifying a storage location for the data within the storage repository.