We are the creators and top committers of Alluxio, formerly Tachyon, and we are revolutionizing the way you store, access, and manage data

Alluxio

T3Go is China’s first platform for smart travel based on the Internet of Vehicles. In this article, Trevor Zhang and Vino Yang from T3Go describe the evolution of their data lake architecture, built on cloud-native or open-source technologies, including Alibaba OSS, Apache Hudi, and Alluxio. Today, their data lake stores petabytes of data, supporting hundreds of pipelines and tens of thousands of tasks daily. It is essential for business units at T3Go, including Data Warehouse, Internet of Vehicles, Order Dispatching, Machine Learning, and self-service query analysis. In this blog, you will see how we slashed data ingestion time by half using Hudi and Alluxio. Furthermore, data analysts using Presto, Hudi, and Alluxio saw the queries speed up by 10 times. We built our data lake based on data orchestration for multiple stages of our data pipeline, including ingestion and analytics. T3Go data lake Overview Prior to the data lake, different business units within T3Go managed their own data processing solutions, utilizing different storage systems, ETL tools, and data processing frameworks. Data for each became siloed from every other unit, significantly increasing cost and complexity. Due to the rapid business expansion of T3Go, this inefficiency became our engineering bottleneck. We moved to a unified data lake solution based on Alibaba OSS, an object store similar to AWS S3, to provide a centralized location to store structured and unstructured data, following the design principles of ; all the applications access OSS storage as the source of truth, as opposed to different data silos. This architecture allows us to store the data as-is without having to first structure the data and run different types of analytics to guide better decisions, building dashboards and visualizations from big data processing, real-time analytics, and machine learning. Multi-cluster Shared-data Architecture Efficient Near Real-time Analytics Using Hudi Our business in smart travel drives the need to process and analyze data in a near real-time manner. With a traditional data warehouse, we faced the following challenges: High overhead when updating due to long-tail latency. High cost of order analysis due to the long window of a business session Reduced query accuracy due to late or ad-hoc updates Unreliability in data ingestion pipeline. Data lost in the distributed data pipeline cannot be reconciled. High latency to access data storage. As a result, we adopted Apache Hudi on top of OSS to address these issues. The following diagram outlines the architecture: Enable near real-time data ingestion and analysis With Hudi, our data lake supports multiple data sources, including Kafka, MySQL binlog, GIS, and other business logs in near real-time. As a result, more than 60% of the company’s data is stored in the data lake, and this proportion continues to increase. We are also able to speed up the data ingestion time down to a few minutes by introducing Apache Hudi into the data pipeline. Combined with big data interactive query and analysis frameworks, such as Presto and SparkSQL, real-time data analysis and insights are achieved. Enable Incremental processing pipeline With the help of Hudi, it is possible to provide incremental changes to the downstream derived table when the upstream table updates frequently. Even with a large number of interdependent tables, we can quickly run partial data updates. This also effectively avoids updating the full partitions of cold tables in the traditional Hive data warehouse. Accessing Data using Hudi as a unified format Traditional data warehouses often deploy Hadoop to store data and provide batch analysis. Kafka is used separately to distribute Hadoop data to other data processing frameworks, resulting in duplicated data. Hudi helps effectively solve this problem; we always use Spark pipelines to insert new updates into the Hudi tables, then incrementally read the update of Hudi tables. In other words, Hudi tables are used as the unified storage format to access data. Efficient Data Caching Using Alluxio In the early version of our data lake without Alluxio, data received from Kafka in real-time is processed by Spark and then written to OSS data lake using Hudi DeltaStreamer tasks. With this architecture, Spark often suffered high network latency when writing to OSS directly. Since all data is in OSS storage, OLAP queries on Hudi data may also be slow due to a lack of data locality. To address the latency issue, we deployed Alluxio as a data orchestration layer, co-located with computing engines such as Spark and Presto, and used Alluxio to accelerate read and write on the data lake as shown in the following diagram: Data in formats such as Hudi, Parquet, ORC, and JSON are stored mostly on OSS, consisting of 95% of the data. Computing engines such as Flink, Spark, Kylin, and Presto are deployed in isolated clusters, respectively. When each engine accesses OSS, Alluxio acts as a virtual distributed storage system to accelerate data, being co-located with each of the computing clusters. Specifically, here are a few applications leveraging Alluxio in the T3Go data lake. Data lake ingestion We mount the corresponding OSS path to the Alluxio file system and set Hudi’s parameter value to use the scheme in place of scheme. Spark pipelines with Hudi continuously ingest data to Alluxio. After data is written to Alluxio, it is asynchronously persisted from the Alluxio cache to the remote OSS every minute. These modifications allow Spark to write to a local Alluxio node instead of writing to remote OSS, significantly reducing the time for the data to be available in the data lake after ingestion. target-base-path alluxio:// oss:// Data analysis on the lake We use Presto as an ad-hoc query engine to analyze the Hudi tables in the lake, co-locating Alluxio workers on each Presto worker node. When Presto and Alluxio services are co-located and running, Alluxio caches the input data locally in the Presto worker, which greatly benefits Presto for subsequent retrievals. On a cache hit, Presto can read from the local Alluxio worker storage at memory speed without any additional data transfer over the network. Concurrent accesses across multiple storage systems In order to ensure the accuracy of training samples, our machine learning team often synchronizes desensitized data in production to an offline machine learning environment. During synchronization, the data flows across multiple file systems, from production OSS to an offline HDFS followed by another offline Machine Learning HDFS. This data migration process is not only inefficient but also error-prone for modelers because multiple different storages with varying configurations are involved. Alluxio helps in this specific scenario by mounting the destination storage systems under the same filesystem to be accessed by their corresponding logical paths in Alluxio namespace. By decoupling the physical storage, allows applications with different APIs to access and transfer data seamlessly. This data access layout also improves performance. Microbenchmark Overall, we observed the following improvements with Alluxio: It supports a hierarchical and transparent caching mechanism. It supports cache promote mode when reading. It supports asynchronous writing mode. It supports LRU recycling strategy. It has pin and TTL features After comparison and verification, we choose to use Spark SQL as the query engine. Our performance testing queries the Hudi table, comparing Alluxio + OSS together against OSS directly as well as HDFS. In the stress test shown above, after the data volume is greater than a certain magnitude (2400W), the query speed using Alluxio+OSS surpasses the HDFS query speed of the hybrid deployment. After the data volume is greater than 1E, the query speed starts to double. After reaching 6E data, it is up to 12 times higher than querying native OSS and 8 times higher than querying native HDFS. The improvement depends on the machine configuration. Based on our performance benchmarking, we found that the performance can be improved by over 10 times with the help of Alluxio. Furthermore, the larger the data scale, the more prominent the performance improvement. Next Step As T3Go’s data lake ecosystem expands, we will continue facing the critical scenario of compute and storage segregation. With T3Go’s growing data processing needs, our team plans to deploy Alluxio on a larger scale to accelerate our data lake storage. In addition to the deployment of Alluxio on the data lake computing engine, which currently is mainly SparkSQL, we plan to add a layer of Alluxio to the OLAP cluster using Apache Kylin and an ad_hoc cluster using Presto. The goal is to have Alluxio cover all computing scenarios, with Alluxio interconnected between each scene to improve the read and write efficiency of the data lake and the surrounding lake ecology. Conclusion As mentioned earlier, Hudi and Alluxio cover all scenarios of Hudi’s near real-time ingestion, near real-time analysis, incremental processing, and data distribution on DFS, among many others, and play the role of a powerful accelerator on data ingestion and data analysis on the lake. With Hudi and Alluxio together, our R&D engineers shortened the time for data ingestion into the lake by up to a factor of 2. Data analysts using Presto, Hudi, and Alluxio in conjunction to query data on the lake saw their queries speed up by 10 times faster. Furthermore, the larger the data scale, the more prominent the performance improvement. Alluxio is an important part of T3Go’s plan to become a leading enterprise data lake in China. We look forward to seeing further integration with Alluxio in T3Go’s data lake ecosystem.

Building High-Performance Data Lake Using Apache Hudi and Alluxio 

Reduce HDFS Latency with Alluxio Managed SSD Cache and Presto Consistent Hashing-Based Soft Affinity Scheduling Background At Uber, data informs every decision. Presto is one of the very core engines that powers all sorts of data analytics at Uber. For example, the operations team makes heavy use of Presto for services such as dashboarding; The Uber Eats and marketing teams rely on the results of these queries to make decisions on prices. In addition, Presto is also used in Uber’s compliance department, growth marketing department, and ad-hoc data analytics. The scale of Presto at Uber is large. Currently, Presto has 7K weekly active users, processing 500K queries per day and handling over 50PB of data. Uber has two data centers with 5000 nodes and 15 Presto clusters across 2 regions in terms of infrastructure. Uber’s Presto Deployment Current Architecture Uber’s Presto architecture is shown in the diagram above. The UI/Client layer. This includes internal dashboarding, Google Data Studio, Tableau, and other tools. In addition, we have some backend services that use JDBC or query parsing to communicate with Presto. The proxy layer. This layer is responsible for pulling stats from each Presto coordinator, including the number of queries and tasks, CPU and Memory usage, and so on. We determine which cluster each query should be scheduled to based on these stats. In other words, it is doing the load-balancing and query-gating services. Presto clusters. At the bottom, multiple Presto clusters communicate with the underlying Hive, HDFS, Pinot, and others. Join operations can be performed between different plugins or different datasets. In addition, for each layer of the above architecture, we have Internal monitoring Support for using Kerberos security The Presto workloads are divided into two categories. Interactive: Query sent by data scientists and engineers Scheduled: mainly Batch query, which is scheduled and recurring, including Dashboard query, ETL query, etc. The Journey to the Cloud Over the past few years, the Uber team has been thinking about how and when to go to the cloud and what the layout should be to interact with the cloud. We are using the “what-why-how” model. we have a variety of application layers, such as BI applications. On the computation engine side, we have Spark, Presto, and so on. In the cloud, there are many cloud-native options. On the storage side, we can choose from GCS, S3, HDFS in the cloud, and many other options. What: One of the most important motivations is to improve cost efficiency and achieve elastic scaling of hardware resources. Also, we want to achieve high usability, scalability, and reliability. Why: On the one hand, because cloud provides many native features, we need to consider features compatibility with open source components. On the other hand, we need to figure out how performance works at different scales. Uber holds a huge data lake, so performance is crucial for the customers and us. Additionally, we are looking into the cloud’s ability to enforce security and compliance requirements. The native cloud-provided services can also be used to reduce “tech debt.” How: Future State – Hybrid Cloud The diagram above shows the long-term plan. The solution can be extended to different cloud service vendors based on such an architecture. Now, we are still in the very early stages of implementation. We want to run a hybrid model with PROD clusters (on-prem) and cloud clusters (GCP). As shown in the bottom right corner of the figure, we expect most of the data to still be on-prem HDFS. The blue arrows represent some pre-tests we did where we ran some experiments without any caching on top of HDFS. The results showed high network traffic and cost, which is a huge overhead. Therefore, we are exploring public cloud services, such as GCS or S3, to serve as an L2 Cache. We want to put some important data or frequently used data sets in the “L2 Cache on the cloud”. For each Presto cluster in the cloud, we plan to leverage local SSDs to cache some data to improve performance. This solution can be scaled into different cloud providers using the same architecture, which is our vision in the long term. Using Alluxio for Local Caching Recently, we deployed Alluxio in our product environment in three clusters, each with more than 200 nodes. Alluxio is plugged in as a local library and it leverages the Presto workers’ local NVMe disks. We’re not caching all the data, but a selective subset of data called selective caching. Below is a diagram of Alluxio as a local cache. The Alluxio Cache Library is a local cache that runs inside the Presto worker. We have implemented a layer on top of the default HDFS client implementation. When any external API reads from an HDFS call, the system first looks at the inventory of the cache to see if it is a cache hit or miss. If it’s a cache hit, it will directly read the data from the local SSD. Otherwise, it will read from the remote HDFS and cache the data locally for the next read. In this process, the cache hit rate has a significant impact on the overall performance. We will discuss the detailed design and improvement of the Alluxio local cache in part 2 of this blog series. Key Challenges and Solution Challenge 1: Real Time Partition Updates The first challenge we encountered was the real time partition updates. At Uber, a lot of tables/partitions are constantly changing because we upsert queries constantly into Hudi tables. The challenge is that the partition id alone as a caching key is not sufficient. The same partition may have changed in Hive, while Alluxio still caches the outdated version. In this case, partitions in cache are outdated so the users will get outdated results while running a query if the data is served from caching, resulting in inconsistent experience. Solution: Add Hive latest modification time to caching key Our solution is adding the latest modification time to the caching key as below: Previous caching key: hdfs://<path> New caching key: hdfs://<path><mod time> Through this solution, the new partition with the latest modification gets cached, ensuring that users always get a consistent view of their data. Note that, there’s a trade-off – outdated partitions still present in cache, wasting caching space until evicted. Currently, we are working to improve the caching eviction strategy. Challenge 2: Cluster Membership Change In Presto, Soft Affinity Scheduling is implemented by the simple mod-based algorithm. This algorithm has the disadvantage that if a node is added or deleted, the whole ring is messed up with a different cache key. Therefore, if a node joins or leaves a cluster, it can hurt the cache efficiency of all the nodes, which is problematic. Presto locates the same set of nodes for a given partition key for this reason. As a result, we always hit the same set of nodes both for caching and queries. While this is good, the problem is that Presto previously used a simple hash function that can break when the cluster changes. As shown below, currently, simple hash mod based node lookup : key 4 % 3 nodes = worker # 1. Now node#3 goes down, new lookup: key 4 % 2 nodes = 0, but worker#0 does not have the bytes. Solution: Node id based consistent hashing Consistent hashing is the solution. Instead of the mod based function, all nodes are placed on a virtual ring. Relative ordering of nodes on the ring doesn’t change, regardless of joining or leaving. Instead of using mode based hash, we always look up the key on the ring. We can ensure that no matter how many changes are made, they are always based on the same set of nodes. Additionally, we use replication to improve robustness. This is the solution to the cluster membership issue. Challenge 3: Cache Size Restriction Uber’s data lake is at large scale, with 50PB of data accumulated per day. However, our local disk space is only 500 GB per node. The amount of data accessed by Presto queries is much larger than the disk space available on Worker nodes. Although it is possible to put everything in the cache, heavy eviction can hurt overall cache performance. Solution: Cache Filter The idea is to only cache a selected subset of data, which includes certain tables and a certain number of partitions. The solution is to develop a cache filter, a mechanism that decides whether to cache a table and how many partitions. Below is a sample configuration: {
  "databases": [{
      "name": "database_foo",
      "tables": [{
          "name" : "table_bar",
          "maxCachedPartitions": 100}]}]
} Having a cache filter has greatly increased cache hit rate . The following are the areas to pay attention to when it comes to Cache Filter. from ~65% to >90% Manual, static configuration Should be based on traffic pattern, e.g.: Most frequently accessed tables Most common # of partitions being accessed Tables that do not change too frequently Ideally, should be based on shadow  caching numbers and table level metrics We’ve also achieved observability through monitoring/dashboarding, which is integrated with Uber’s internal metrics platform using Jmx metrics emitted to Grafana based dashboard. Current Status and Future Work Current Status We have deployed 3 clusters of 200+ nodes each to our production cluster, with all nodes on NVMe disks, and 500GB cache space per node. We used a cache filter to cache about 20 most frequently accessed tables. We are still in the early stage of deployment and monitoring results. The initial measurement shows great improvement with about 1/3 of wall time for input scan (TableScanOperator and ScanFilterProjectOperator) vs. no cache.  We are very encouraged by these results. Alluxio has shown great query acceleration and provides consistent performance in our big data processing use cases. Next Steps First, we would like to onboard more tables and improve the process of table onboarding with automation, in which Alluxio Shadow Cache (SC) will be helpful. Second, we want to have better support for the changing partitions/Hoodie tables. Lastly, load balancing is another optimization we can implement. There’s still a long way to go along our journey. As compute-storage separation continues to be the trend along with containerization in big data, we believe a unified layer that bridges the compute and storage like Alluxio will continue to play a key role. We will share the detailed design and improvement of the Alluxio local cache in the next blog. About the Authors Chen Liang Chen Liang is a senior software engineer at Uber’s interactive analytics team, focusing on Presto. Before joining Uber, Chen was a staff software engineer at LinkedIn’s Big Data platform. Chen is also a committer and PMC member of Apache Hadoop. Chen holds two master’s degrees from Duke University and Brown University. Beinan Wang Dr. Beinan Wang is a software engineer from Alluxio and is the committer of PrestoDB. Before Alluxio, he was the Tech Lead of the Presto team in Twitter, and he built large-scale distributed SQL systems for Twitter’s data platform. He has twelve-year experience working on performance optimization, distributed caching, and volume data processing. He received his Ph.D. in computer engineering from Syracuse University on the symbolic model checking and runtime verification of distributed systems.

Speeding Up Uber’s Presto with Alluxio - Part 1

Written by:  Tyler Crain Introduction Alluxio implements a virtual distributed file system that allows accessing independent large data stores with compute engines like Hadoop or Spark through a single interface. Within an Alluxio cluster, the Alluxio master is responsible for coordinating and keeping access to the underlying data stores or filesystems (UFS for short) consistent. The master contains a global snapshot of filesystem metadata, and when a client wants to read or modify a file, it first contacts the master. Given its central role in the system, the master must be fault-tolerant, highly available, strongly consistent, and fast. This blog will discuss the evolution of the Alluxio master from a complex multi-component system using Zookeeper to a simpler and more efficient one using Raft. The operation of a file system can be thought of as a sequence of commands performed on the system (e.g., create/delete/read/write). Executing these commands one at a time in a single thread gives us a sequential specification for our file system that is easy to reason about and to implement applications on top of. While this sequential specification is simple, the actual implementation of Alluxio’s virtual distributed file system consists of many moving parts all executing concurrently: the master who coordinates the system, the workers who store the file data and act as a cache between the client and the UFS, the UFSs themselves, and the clients who, at the direction of the master, access all other parts of the system. Again here, we may want to reason about each of these individual components as a sequential thread executing one operation at a time, but in reality, they are running complex systems themselves. The core problem we want to solve here is how to make the Alluxio master seem like it is being run by a single node running a single operation at a time so that it is simple to implement and easy for other components to interact with, yet still be fault-tolerant and highly available. Furthermore the master must be fast enough to perform tens of? thousands of operations per second and scale to support the metadata of hundreds of? millions of files. The Journal and HA (Highly Availability) The internal state of the Alluxio master is made up of several tables containing the filesystem metadata, the block metadata, and the worker metadata. The filesystem metadata consists of an inode table that maps each directory and file’s metadata (its name, parent, etc) to a unique identifier called the inode id, and an edge table that maps directories to their children using the inode ids. The actual data of each file is split into one or more blocks located on the Alluxio workers which can be located using the block metadata table which maps each file to a set of block identifiers and the worker metadata table which maps each worker to the list of blocks it contains. Each operation that modifies this metadata, for example creating a new file on the virtual file system, outputs a containing the modifications made to the tables. The journal entries are stored in the order they are created in a log, thus satisfying our sequential specification of the master. Starting from an initial state and replaying these journal entries in order will result in the same state of the metadata, allowing a master node to recover after a failure. To ensure the journal does not grow too large, snapshots of the metadata will be taken periodically. Now if this journal is stored in a location that is highly available and fault-tolerant, then we have a basis for our reliable system. For example, we can start a single node and assign it the role of master. Some time later, if we detect that the node has failed, then a new node is started, it replays the journal, and takes its place as the new master. One possible issue with this is that it may take a long time to start a new node and replay the journal, during which time the system is unavailable. To increase the availability of the system, we can have multiple replicas of the master running at the same time where only one is serving client operations (call this master the ), while the others are just replaying journal entries as they are created (call these masters the ). Now when the primary master fails, a secondary master can take over immediately as the new primary master. journal entry primary master secondary masters Implementing Our Initial Design From this basic design, there remain two problems to solve, 1: How to ensure there is a single primary master running at all times, and 2: how to ensure the journal itself is consistent, fault tolerant, and highly available. Starting in 2013, we introduced the initial implementation to resolve these two problems. The first is solved by Zookeeper [1]. Zookeeper is a highly reliable distributed coordination service providing a filesystem-like interface running over the ZAB [2] consensus algorithm. Zookeeper provides a leader election [3] where multiple nodes can join the recipe and a single node will be chosen as the leader. If a leader is no longer responsive for a certain period then a new leader will be chosen. In Alluxio all masters start as secondary masters and join the Zookeeper leader election recipe, when the recipe elects a node as leader it can then to primary master. recipe step-up To solve the second problem and ensure the journal is highly available and fault-tolerant, the journal entries are stored on the UFS. Of course a UFS that is available and fault-tolerant should be used. Finally, the journal must be consistent. As described earlier the journal entries should be a totally ordered log of events satisfying our sequential specification. As long as there is no more than one primary master running at a time it can just append journal entries to a file in the UFS to satisfy consistency. But doesn’t our leader election recipe already ensure this? If we have a perfect failure detector or a synchronous system, then yes, but unfortunately we have neither. When Zookeeper performs a leader election it will notify the previous leader it has lost its leadership before electing the new leader, but if the previous leader (or the network) is slow, it may still be appending entries to the journal when the new leader is elected. As a result, we may have multiple primary masters appending to the journal concurrently. To reduce the chance of this happening, a mutual exclusion protocol is run using the consistency guarantees of the UFS by using a specific file naming scheme. Let us examine some of the issues of this solution. First, it relies on several external systems, Zookeeper and a UFS, running alongside the Alluxio masters. This complicates the design and analysis of the system. Each component has a different fault and availability model to consider. The choices between different UFSs also complicate the system, for example, one UFS may not provide high performance with frequent small append operations. Furthermore, each UFS may have different consistency guarantees, complicating the case when two concurrent Primary Masters are trying to write to the journal. For these reasons, HDFS is the recommended UFS for this configuration. We now describe how a simpler and more efficient design can be made using Raft. An Improved Design Based on the Replicated State Machine First let us introduce the concept of a ( . A state machine takes as input a series of commands which modify the internal state of the system and may produce some outputs. For our case, the state machine must be deterministic, namely given the same set of inputs, the system will produce the same outputs. If we think of our journal as a state machine, we can think of it as simply a log which has an append command that appends a journal entry to the end of the log. A is then a state machine that provides high availability and fault tolerance. It does this by replicating the state machine over multiple nodes and running a consensus algorithm to ensure the same commands are run in the same order at each replica. Raft [4] is one such replicated state machine protocol, a user just needs to provide a deterministic state machine as input to the protocol. One important property about Raft is that it ensures the consistency condition of [5]. Consider that each command executed on the replicated state machine by a client has an invocation and response. Linearizability ensures that an operation is observed by all clients to have taken place at some instant in time between the invocation and response. This instant in time is called the operation’s . The linearization points of all operations give us a total order of the operations following the sequential specification of the state machine being implemented. So even though our state machine is replicated, to the client it appears to be executing operations sequentially in real-time. replicated) state machine replicated state machine linearizability linearization point Implementing the Journal using Raft Now we could simply provide the log of journal entries as the state machine to be replicated by Raft, but we can simplify the system further. Recall that the reason for representing the state as a log of journal entries was to allow masters to recover to the most recent state by replaying the journal entries. Since Raft takes care of fault tolerance, we can use a higher level of abstraction for our state machine that represents the state of the master. This is simply the collection of tables containing the state of the metadata of the Alluxio file system. The master will execute its normal operations, and provide to the statemachine commands that represent modifications to the tables, which are clearly deterministic. Raft internally handles the logging, snapshotting and recovery of the state in case of failures. The Raft code runs alongside the Alluxio code on the master nodes allowing quick access to the metadata in the tables stores. The tables are implemented using RocksDB [6]. RocksDB is an on-disk key-value store implemented using a log-structured merge-tree [7] which allows for efficient updates. By using RocksDB the size of the filesystem metadata can grow larger than the memory available on the master nodes. Alluxio additionally implements an in memory cache above RocksDB to ensure reading keys is fast. Like before, a single master node will be designated the primary master which serves the client requests. This is because the virtual file system operations that create the journal modifications should only happen at the primary master as they may modify the external state (the UFS and worker nodes). Raft internally uses leader election as part of its consensus algorithm which Alluxio uses to appoint a primary master. To decrease the likelihood of two primary masters running concurrently, Alluxio adds an additional layer of synchronization allowing the previous primary master to explicitly step down or timeout before the new primary master starts serving client requests. Using Raft has greatly simplified the implementation and improved the performance and scalability of the highly available and fault-tolerant masters of Alluxio. The running of the external systems of Zookeeper and a UFS has been replaced with the Apache Ratis [8] implementation of Raft running directly on the Alluxio master nodes. The concept of a low-level log to be replayed after recovery is no longer needed. Instead, we only need to access fast key-value stores using RocksDB, which are efficiently and consistently replicated by Raft to ensure high availability and fault tolerance. We also recommend reading how Confluent simplified and improved the performance of their distributed architecture by moving from Zookeeper to Raft in Apache Kafka [9]. References [1] Hunt, Patrick, et al. “{ZooKeeper}: Wait-free Coordination for Internet-scale Systems.” 2010 USENIX Annual Technical Conference (USENIX ATC 10). 2010. . https://zookeeper.apache.org [2] Junqueira, Flavio P., Benjamin C. Reed, and Marco Serafini. “Zab: High-performance broadcast for primary-backup systems.” 2011 IEEE/IFIP 41st International Conference on Dependable Systems & Networks (DSN). IEEE, 2011. [3] Apache Curator Leader Election. https://curator.apache.org/curator-recipes/leader-election.html [4] Ongaro, Diego, and John Ousterhout. “In search of an understandable consensus algorithm.” 2014 USENIX Annual Technical Conference (Usenix ATC 14). 2014. . https://raft.github.io [5] Herlihy, Maurice P., and Jeannette M. Wing. “Linearizability: A correctness condition for concurrent objects.” ACM Transactions on Programming Languages and Systems (TOPLAS) 12.3 (1990): 463-492. [6] RocksDB. . http://rocksdb.org [7] O’Neil, Patrick, et al. “The log-structured merge-tree (LSM-tree).” Acta Informatica 33.4 (1996): 351-385. [8] Apache Ratis. .[9] Apache Kafka Made Simple: A First Glimpse of a Kafka Without ZooKeeper. . https://ratis.apache.org https://www.confluent.io/blog/kafka-without-zookeeper-a-sneak-peek/ Also Published Here

How Alluxio Stores File System State with High Availability, Fault Tolerance

This is a guest blog contributed by datasapiens’ Juraj Pohanka, Koen Michiels and Sam Gilbert. This article described how engineers at datasapiens brought down S3 API costs by 200x by implementing Alluxio as a data orchestration layer between S3 and Presto. . I Introduction Previous works There have been numerous articles and online webinars dealing with the benefits of using Alluxio as an between the S3 data storage and the data processing system used for ingestion or retrieval of data (i.e. Spark, Presto), as depicted in the picture below: intermediate storage layer To name a few use cases Alluxio-Presto use cases: Ryte ING Alluxio-Spark use cases: e Bazaarvoic The main conclusion from these use cases is that using Alluxio as a data orchestration layer has the following benefits: in data processing pipelines: lower latency Co-locating data and computation reduces network traffic for usage concurrency: horizontal scalability S3 API has limitations on the request rate for a given prefix All these benefits are critical when deciding a production-grade data processing architecture, however one important benefit has so far not been sufficiently highlighted when choosing such architecture – on the usage of the S3 API. This will be the focus of this article. cost reduction datasapiens is an international startup, based in Prague, that focuses primarily on helping companies to transform their business through data. We offer 3 key products: datasapiens our business intelligence platform with an embedded insights framework to help clients to become data-driven companies. Business insights commercialize clients’ data assets by driving cooperation with their suppliers. Brands insights increase customer engagement, sales and margin through a simple plan-do-review process that continuously measures and improves the loyalty lifecycle of clients’ customers. Personalized offers Our infrastructure journey of our infrastructure used AWS Redshift as the main data warehouse. However, when we started to acquire clients with larger data volumes (i.e. billions of records per year) we started to struggle with the following: The first generation storage-compute coupled architecture large infrastructure costs long query compilation times vendor-locking It was at this point that we decided to move to the . Hadoop ecosystem was the common software stack for cloud-based data lake architectures: The first version of our second generation S3 for data persistence Spark for ETL Hive for data warehousing Presto as a distributed query engine With this storage-compute separated architecture, we achieved faster performance and reduced compilation times. However, the infrastructure costs were above expectations due to S3 API cost increases, making up a large proportion of the overall cost (described in detail below). includes Alluxio as a data orchestration layer. We co-locate our data with our compute for in-memory data access. This architecture achieves even faster performance more consistently. In addition, we saw a drop in infrastructure costs due to S3 API costs dropping to a negligible level. The second version of our second generation . II Our observations A drastic increase in S3 API costs During the fine-tuning of our first Presto cluster, we conducted several performance tests of the cluster under various levels of user concurrency. The setup of the performance tests was the following: : dataset setup internal client data (1 billion rows of retail transactional data) persisted in Parquet workload setup 23 different queries used in client’s reports concurrency levels from 1 to 150 Ryte : cluster hardware setup worker nodes: count: 10 type: c5.4xlarge master nodes: count: 1 type: c5.4xlarge cluster software deployment setup PrestoDB 0.227 Apache Hive v2.3.5 For the design of our DWH, we wanted to avoid using the traditional star/snowflake schemas. Instead, we could pre-join the table entities in our ETL into a few OLAP cubes to avoid expensive table joins in Presto. Towards the end of the testing phase, we detected a dramatic increase in costs for the S3 API service. The following image depicts the incurred S3 API costs for each day of the performance tests: The following table summarizes the total number of executed queries with the S3 API costs and S3 API request counts per each day of performance tests: The next table shows the percentage of the S3 API costs out of the total costs: During this testing period, we tested several configuration variations of the Presto cluster in order to achieve a satisfying query speed at a specific concurrency level. In the first table above, we can observe the variance in the S3 API costs/requests per day. We attribute this variance to the varying Presto cluster deployments for each day in the testing period. After reviewing the incurred S3 API costs under the given workloads, we came to the conclusion that running a Presto cluster that read directly from S3 was not feasible, as it would impact our profit margin. Alluxio as a remedy We considered deploying Hadoop as an intermediate layer between Presto and S3 and synchronizing through simple jobs. We also looked at Alluxio, which provided similar functionality without the need for manual synchronization. Furthermore, Alluxio has additional benefits such as connections to various cloud providers & storage devices and having tiered storage functionality which we thought would improve query speed. After deploying a co-located Alluxio-Presto cluster, the performance of queries improved (especially for those with a higher number of execution stages), and the S3 API costs dropped to below $1/day. We have since implemented co-located Alluxio-Presto clusters in production. The infrastructure is successfully serving hundreds of users per client. We also learned that although manual synchronization is not needed. To prevent any performance degradation of the Presto clusters during startup and after ETL jobs it’s best to “hydrate” the cluster using “warm-up” jobs. . III A reproducible example using TPC-DS To get a better quantitative understanding of the S3 API costs and the cost reduction when using Alluxio, we conducted a simple experiment by querying data from S3 in two separate clusters: a co-located Alluxio-Presto cluster using PrestoSQL deployed in a Docker Swarm cluster an EMR cluster with PrestoDB without Alluxio In both clusters, we measured the API request count to Alluxio per each query. The following are the setup schemas for each cluster: Dataset setup: TPC-DS dataset with scale factor 100 stored in a S3 bucket Alluxio-Presto cluster setup: hardware setup worker nodes: count: 6 type: c5.4xlarge master nodes: count: 1 type: m5.2xlarge software deployment setup vertions of used systems: Alluxio 2.2.0 PrestoSQL 337 EMR-Presto cluster setup: hardware setup worker nodes: count: 6 type: c5.4xlarge master nodes: count: 1 type: m5.2xlarge software deployment setup PrestoDB 0.227 Query execution setup: set of queries: TPC-DS suite excluding Query no. 72 (due to longer query times) query executions: number of repeats: 10 concurrency level: 1 (no parallel executions) Measurements setup: Alluxio: logical Operations: ‘File Infos Got’ RPC Invocations: ‘GetFileInfo Operations’ S3: total request counts per request type total request costs per request type The deployment configuration for each of the clusters, as well as the complete test results, can be found in our . GitHub repository Here are the results from the conducted tests on the Alluxio-Presto cluster: 10 most API request-expensive queries: cumulative request counts: S3 API costs for caching the dataset into Alluxio: Here are the results from the conducted tests on the EMR-Presto cluster: As we can observe, the cumulative count for the `GetObject` request type and the `File Infos Got` request type are more-less compatible. The different values for the storage API calls in both cases can be attributed to: different nature of the Alluxio and S3 file system API different versions of used Presto clusters (PrestoSQL 337 vs PrestoDB 0.227) In the case of the S3 API, we also have an additional cost for the `ListBucket` request type which is more costly then the `GetObject` type. Now, we will try to estimate the per-query S3 API costs for the EMR-Presto cluster. We are limited by the following facts: we cannot tag individual requests to S3 we do not know the exact number of each API request type incurred by individual query To estimate the per-query S3 API costs, we will take a naïve approach and redistribute the costs to each query proportionally to the number of requests sent to Alluxio. This gives us the following table with the top 10 request-expensive queries: As can be seen, we can have quite significant costs for a single executed query. Let's do a simple comparison of the cluster infrastructure and S3 API costs for both cases with and without Alluxio: It is apparent that the S3 API costs for the cluster with Alluxio form a negligible (0.58%) part of the total costs, whereas for the cluster without Alluxio, they form almost half (48.83%) of the total costs in our simple test case. . IV General implications for cloud data lake architectures Looking at the pricing models for the storage services of three largest cloud providers ( , , and ), we see that they are very similar to each other, hence one can expect similar storage API costs when running analytical workloads with or without Alluxio. However, this claim still needs to be verified. AWS S3 Azure Blob Store Google Cloud Storage Our test case was a simple setup of a small cluster to which we were sending queries in a sequential manner. However, real-life production environments cope with workloads with far higher compute and storage demands. Frequently, we have: a large number of processes and users frequently and concurrently running queries workloads running real-time data streaming, ingesting and retrieving large amounts of data For these types of cloud-based production environments, using no intermediate data storage layer between the cloud storage service and the data processing system can lead to tremendous costs. Therefore we would be inclined to recommend that when running analytical workloads in a cloud-based environment, to have a data orchestration layer such as Alluxio. The original article was published on Alluxio’s Engineering Blog .

How To Optimize Large S3 API Costs using Alluxio

This article describes how engineers in the Data Service Center (DSC) at Tencent PCG (Platform and Content Business Group) leverages Alluxio to optimize the analytics performance and minimize the operating costs in building , a real-time data analytics platform. Tencent Beacon Growing Specifically, by using Alluxio as a distributed caching layer, the HDFS scanning performance for Impala SQL queries is accelerated by 2x out of the box without further optimizations. The original article was published on by guest writer Honghan Tian. Alluxio's Engineering Blog Tencent’s Big Data Analytic App Beacon Growing is an analytics platform that provides analyzers insights of user behaviors in realtime. Beacon comes with both built-in and customized analytic models, such as RETENT, FUNNEL, PATH, etc. Combining the functions of A/B testing and user profiles, developers can select promotional plans more easily. Under the hood, Beacon is powered by a stack of open source technologies. Particularly for the context of this article, we used Impala to query data in Parquet format stored in HDFS. Here is the overview of the software stack currently being deployed by Beacon, to give the readers a better understanding of the system. Challenges Before In Tencent, we build systems to handle data at the Internet-scale. For example, the scan range of a typical query can be 100 billion rows, with hundreds of concurrent queries. To speed up the queries, we have applied optimizations like . In practice, there were still a few issues observed: page index in Parquet files to reduce the I/O cost in scan operations : When more than 100 queries are running concurrently and scanning HDFS, we observed that the usage of host disk was fixed at 100%. I/O Bottleneck under high concurrency :  Insert operations such as “SORT BY” during table creation are very slow due to files merge under HDFS-parquet sort. Note that in our , “SORT BY” helps meet query SLAs without over partitioning the table. Expensive writes application Why Alluxio Based on the above observations, we decided to build a caching layer using SSDs. However, HDFS is not designed to leverage heterogeneous storage. We deployed Alluxio to leverage the speed advantage of SSDs and the capacity advantage of HDDs. The new architecture is shown in the following figure: There are a few changes and improvements observed after integrating Alluxio into Tencent Beacon infrastructure: The range of the average load of HDFS was significantly reduced from  80% ~ 100% down to 50% ~ 70%. The I/O time is also reduced by 50% or more. Better Resource Utilization: : With faster data retrieval, I/O-intensive queries are sped up by 244% with a median improvement of 121% across all queries. Higher Query Performance Alluxio greatly reduces the I/O consumption of hotspot and large queries to the entire HDFS cluster, reducing the overall query failure rate by more than 5% Higher Cluster Stability: Faster execution also reduces the query time in waiting-queue, resulting in 29% reduction in the timeout query failure rate. Reduced Timeout Failure Rate: One of the most valuable features of Alluxio in our use case is . It helps share the same namespace across multiple clusters, making it easier for Impala to access the data resources of multiple HDFS clusters. Unified Namespace The following table is a comparison of some data： Next steps: Locality Groups There are a few on-going projects on the horizon to integrate Alluxio at Tencent PCG. First, we plan to further reduce the operating work by deploying a large Alluxio cluster with worker processes spanning multiple Impala clusters. In other words, Alluxio and Impala processes will be co-located at the same server to reduce network traffic. We are currently experiencing some and working with the core team to address them. issues Second, we want to leverage the Alluxio to further improve Impala performance and eliminate remote reads of Alluxio workers with improved data locality. tiered locality group Specifically, with one but much larger Alluxio cluster deployed, we will divide Alluxio workers nodes into different ‘locality groups’. As a result, when queries read data from different HDFS clusters, data will be cached into the nearest locality group to reduce I/O cost. This feature is an analogy of the in Clickhouse. “layer” configuration” Third, once locality grouping described above is achieved, locally deployed Impala computing groups can be used to elastically expand or shrink computing resources based on different Impala clusters for different business uses. Conclusion In our project “Beacon Growing”, we have deployed Alluxio to improve Impala performance by 2.44x for IO intensive queries and 1.20x for all queries. The query failure rate due to timeout is also reduced by 29%. In the future, we foresee it can reduce disk utilization by over 20% for our planned elastic computing on Impala.

Accelerating Analytics by 200% with Impala, Alluxio, and HDFS at Tencent

This tutorial shows how Alibaba Cloud Container team runs PyTorch on HDFS using Alluxio under Kubernetes environment. The original Chinese article was published on , then translated and published on Alibaba Cloud's engineering blog Alluxio's Engineering Blog Google’s TensorFlow and Facebook’s PyTorch are two Deep Learning frameworks that have been popular with the open source community. Although PyTorch is still a relatively new framework, many developers have successfully adopted it due to its ease of use. By default, PyTorch does not support Deep Learning model training directly in HDFS, which brings challenges to users who store data sets in HDFS. These users need to either export HDFS data at the start of each training job or modify the source code of PyTorch to support reading from HDFS. Both approaches are not ideal because they require additional manual work that may introduce additional uncertainties to the training job. To avoid this problem, we choose to use Alluxio as an interface to access  HDFS via a POSIX FileSystem interface. This approach greatly improved development efficiency at Alibaba Cloud. This article demonstrates how this work was achieved within a Kubernetes environment. Prepare HDFS 2.7.2 environment For this tutorial, we used a Helm Chart to install HDFS to mock an existing HDFS cluster. 1. Install the helm chart of Hadoop 2.7.2 git https://github.com/cheyang/kubernetes-HDFS.git

kubectl label nodes cn-huhehaote.192.168.0.117 hdfs-namenode-selector=hdfs-namenode-0 helm dependency build charts/hdfs-k8s
helm install hdfs charts/hdfs-k8s \
      -- tags.ha= \
      -- tags.simple= \
      -- global.namenodeHAEnabled= \
      -- hdfs-simple-namenode-k8s.nodeSelector.hdfs-namenode-selector=hdfs-namenode-0 clone #helm install -f values.yaml hdfs charts/hdfs-k8s set false set true set false set 2. Check the status of the helm chart kubectl get all -l release=hdfs 3. Client access hdfs kubectl -it hdfs-client-f5bc448dd-rc28d bash
root@hdfs-client-f5bc448dd-rc28d:/ Configured Capacity: 422481862656 (393.47 GB)
Present Capacity: 355748564992 (331.32 GB)
DFS Remaining: 355748515840 (331.32 GB)
DFS Used: 49152 (48 KB)
DFS Used%: 0.00%
Under replicated blocks: 0
Blocks with corrupt replicas: 0
Missing blocks: 0
Missing blocks (with replication factor 1): 0

-------------------------------------------------
Live datanodes (2):

Name: 172.31.136.180:50010 (172-31-136-180.node-exporter.arms-prom.svc.cluster.local)
Hostname: iZj6c7rzs9xaeczn47omzcZ
Decommission Status : Normal
Configured Capacity: 211240931328 (196.73 GB)
DFS Used: 24576 (24 KB)
Non DFS Used: 32051716096 (29.85 GB)
DFS Remaining: 179189190656 (166.88 GB)
DFS Used%: 0.00%
DFS Remaining%: 84.83%
Configured Cache Capacity: 0 (0 B)
Cache Used: 0 (0 B)
Cache Remaining: 0 (0 B)
Cache Used%: 100.00%
Cache Remaining%: 0.00%
Xceivers: 1
Last contact: Tue Mar 31 16:48:52 UTC 2020 exec # hdfs dfsadmin -report 4. HDFS client configuration [root@iZj6c61fdnjcrcrc2sevsfZ kubernetes-HDFS] root@hdfs-client-f5bc448dd-rc28d:/ cat: /etc/hadoop-custom-conf: Is a directory
root@hdfs-client-f5bc448dd-rc28d:/ root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf core-site.xml  hdfs-site.xml
root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf <?xml version= ?>
<?xml-stylesheet = href= ?>
<configuration>
  <property>
    <name>fs.defaultFS</name>
    <value>hdfs://hdfs-namenode-0.hdfs-namenode.default.svc.cluster.local:8020</value>
  </property>
</configuration>
root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf <?xml version= ?>
<?xml-stylesheet = href= ?>
<configuration>
  <property>
    <name>dfs.namenode.name.dir</name>
    <value>file:///hadoop/dfs/name</value>
  </property>
  <property>
    <name>dfs.namenode.datanode.registration.ip-hostname-check</name>
    <value> </value>
  </property>
  <property>
    <name>dfs.datanode.data.dir</name>
    <value>/hadoop/dfs/data/0</value>
  </property>
</configuration>
root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf Error: No named `--version root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf Error: No named `-version root@hdfs-client-f5bc448dd-rc28d:/etc/hadoop-custom-conf Hadoop 2.7.2
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r b165c4fe8a74265c792ce23f546c64604acf0e41
Compiled by jenkins on 2016-01-26T00:08Z
Compiled with protoc 2.5.0
From with checksum d0fda26633fa762bff87ec759ebe689c
This was run using /opt/hadoop-2.7.2/share/hadoop/common/hadoop-common-2.7.2.jar # kubectl exec -it hdfs-client-f5bc448dd-rc28d bash # cat /etc/hadoop-custom-conf # cd /etc/hadoop-custom-conf # ls # cat core-site.xml "1.0" type "text/xsl" "configuration.xsl" # cat hdfs-site.xml "1.0" type "text/xsl" "configuration.xsl" false # hadoop --version command ' was found. Perhaps you meant `hadoop -version' # hadoop -version command ' was found. Perhaps you meant `hadoop version' # hadoop version source command 5. Experimental HDFS basic file operations Found 1 items
drwxr-xr-x   - root supergroup          0 2020-03-31 16:51 / Found 2 items
-rw-r--r--   3 root supergroup       3670 2020-04-20 08:51 / /hadoop-env.cmd # hdfs dfs -ls / test # hdfs dfs -mkdir /mytest # hdfs dfs -copyFromLocal /etc/hadoop/hadoop-env.cmd /test/ # hdfs dfs -ls /test test 6. Download data mkdir -p /data/MNIST/raw/ /data/MNIST/raw/
wget http://kubeflow.oss-cn-beijing.aliyuncs.com/mnist/train-images-idx3-ubyte.gz
wget http://kubeflow.oss-cn-beijing.aliyuncs.com/mnist/train-labels-idx1-ubyte.gz
wget http://kubeflow.oss-cn-beijing.aliyuncs.com/mnist/t10k-images-idx3-ubyte.gz
wget http://kubeflow.oss-cn-beijing.aliyuncs.com/mnist/t10k-labels-idx1-ubyte.gz
hdfs dfs -mkdir -p /data/MNIST/raw
hdfs dfs -copyFromLocal *.gz /data/MNIST/raw cd Deploy Alluxio 1. First select the designated node, which can be one or more kubectl label nodes cn-huhehaote.192.168.0.117 dataset=mnist 2. Create config.yaml, in which you need to configure the node selector to specify the node cat << EOF > config.yaml
image: registry.cn-huhehaote.aliyuncs.com/alluxio/alluxio
imageTag: nodeSelector:
    dataset: mnist
properties:
    alluxio.fuse.debug.enabled: alluxio.user.file.writetype.default: MUST_CACHE
    alluxio.master.journal.folder: /journal
    alluxio.master.journal.type: UFS
    alluxio.master.mount.table.root.ufs: worker:
    jvmOptions: master:
    jvmOptions: tieredstore:
  levels:
  - : MEM
    level: 0
    quota: 20GB : hostPath
    path: /dev/shm
    high: 0.99
    low: 0.8
fuse:
  image: registry.cn-huhehaote.aliyuncs.com/alluxio/alluxio-fuse
  imageTag: jvmOptions: args:
    - fuse
    - --fuse-opts=direct_io
EOF "2.2.0-SNAPSHOT-b2c7e50" "false" "hdfs://hdfs-namenode-0.hdfs-namenode.default.svc.cluster.local:8020" " -Xmx4G " " -Xmx4G " alias type "2.2.0-SNAPSHOT-b2c7e50" " -Xmx4G -Xms4G " It should be noted that theHDFS version needs to be specified at compilation time. This example provides a container image that supports HDFS 2.7.2 is used to specify the HDFS address alluxio.master.mount.table.root.ufs quota is used to specify the upper limit of the cache. For specific configuration, please refer to https://docs.alluxio.io/os/user/stable/en/deploy/Running-Alluxio-On-Kubernetes.html 3. Deploy Alluxio wget http://kubeflow.oss-cn-beijing.aliyuncs.com/alluxio-0.12.0.tgz
tar -xvf alluxio-0.12.0.tgz
helm install alluxio -f config.yaml alluxio 4. Check the status of Alluxio, wait until all components are ready helm get manifest alluxio | kubectl get -f -
NAME                     TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)                                   AGE
service/alluxio-master   ClusterIP   None         <none>        19998/TCP,19999/TCP,20001/TCP,20002/TCP   14h

NAME                            DESIRED   CURRENT   READY   UP-TO-DATE   AVAILABLE   NODE SELECTOR   AGE
daemonset.apps/alluxio-fuse     4         4         4       4            4           <none>          14h
NAME                            DESIRED   CURRENT   READY   UP-TO-DATE   AVAILABLE   NODE SELECTOR   AGE
daemonset.apps/alluxio-worker   4         4         4       4            4           <none>          14h

NAME                              READY   AGE
statefulset.apps/alluxio-master   1/1     14h Prepare PyTorch container image 1. Create a Dockerfile mkdir pytorch-mnist pytorch-mnist
vim Dockerfile cd Populate the Dockerfile with the following content: FROM pytorch/pytorch:1.4-cuda10.1-cudnn7-devel ADD mnist.py /

CMD [ , ] # pytorch/pytorch:1.4-cuda10.1-cudnn7-devel "python" "/mnist.py" 2. Create a PyTorch python file called mnist.py pytorch-mnist
vim mnist.py cd Populate the python file with the following content: __future__ print_function argparse torch torch.nn nn torch.nn.functional F torch.optim optim torchvision datasets, transforms torch.optim.lr_scheduler StepLR super(Net, self).__init__()
        self.conv1 = nn.Conv2d( , , , )
        self.conv2 = nn.Conv2d( , , , )
        self.dropout1 = nn.Dropout2d( )
        self.dropout2 = nn.Dropout2d( )
        self.fc1 = nn.Linear( , )
        self.fc2 = nn.Linear( , ) x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = F.max_pool2d(x, )
        x = self.dropout1(x)
        x = torch.flatten(x, )
        x = self.fc1(x)
        x = F.relu(x)
        x = self.dropout2(x)
        x = self.fc2(x)
        output = F.log_softmax(x, dim= ) output model.train() batch_idx, (data, target) enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = F.nll_loss(output, target)
        loss.backward()
        optimizer.step() batch_idx % args.log_interval == :
            print( .format(
                epoch, batch_idx * len(data), len(train_loader.dataset), * batch_idx / len(train_loader), loss.item())) model.eval()
    test_loss = correct = torch.no_grad(): data, target test_loader:
            data, target = data.to(device), target.to(device)
            output = model(data)
            test_loss += F.nll_loss(output,
                                    target,
                                    reduction= ).item()
            pred = output.argmax(dim= , keepdim= )
            correct += pred.eq(target.view_as(pred)).sum().item()

    test_loss /= len(test_loader.dataset)

    print( .format(
        test_loss, correct, len(test_loader.dataset), * correct / len(test_loader.dataset))) parser = argparse.ArgumentParser(description= )
    parser.add_argument( , type=int, default= , metavar= ,
                        help= )
    parser.add_argument( , type=int,
                        default= ,
                        metavar= ,
                        help= )
    parser.add_argument( , type=int, default= , metavar= ,
                        help= )
    parser.add_argument( , type=float, default= , metavar= ,
                        help= )
    parser.add_argument( , type=float, default= , metavar= ,
                        help= )
    parser.add_argument( , action= , default= ,
                        help= )
    parser.add_argument( , type=int, default= , metavar= ,
                        help= )
    parser.add_argument( , type=int, default= , metavar= ,
                        help= )

    parser.add_argument( , action= , default= ,
                        help= )
    args = parser.parse_args()
    use_cuda = args.no_cuda torch.cuda.is_available()

    torch.manual_seed(args.seed)

    device = torch.device( use_cuda )

    kwargs = { : , : } use_cuda {}
    train_loader = torch.utils.data.DataLoader(
        datasets.MNIST( , train= , download= ,
                       transform=transforms.Compose([
                           transforms.ToTensor(),
                           transforms.Normalize(( ,), ( ,))
                       ])),
        batch_size=args.batch_size, shuffle= , **kwargs)
    test_loader = torch.utils.data.DataLoader(
        datasets.MNIST( , train= , transform=transforms.Compose([
                       transforms.ToTensor(),
                       transforms.Normalize(( ,), ( ,))
                       ])),
        batch_size=args.test_batch_size, shuffle= , **kwargs)

    model = Net().to(device)
    optimizer = optim.Adadelta(model.parameters(), lr=args.lr)

    scheduler = StepLR(optimizer, step_size= , gamma=args.gamma) epoch range( , args.epochs + ):
        train(args, model, device, train_loader, optimizer, epoch)
        test(model, device, test_loader)
        scheduler.step() args.save_model:
        torch.save(model.state_dict(), ) __name__ == :
    main() # -*- coding: utf-8 -*- # @Author: cheyang # @Date:   2020-04-18 22:41:12 # @Last Modified by:   cheyang # @Last Modified time: 2020-04-18 22:44:06 from import import import import as import as import as from import from import : class Net (nn.Module) : def __init__ (self) 1 32 3 1 32 64 3 1 0.25 0.5 9216 128 128 10 : def forward (self, x) 2 1 1 return : def train (args, model, device, train_loader, optimizer, epoch) for in if 0 'Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}' 100. : def test (model, device, test_loader) 0 0 with for in 'sum' 1 True '\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n' 100. : def main () # Training settings 'PyTorch MNIST Example' '--batch-size' 64 'N' 'input batch size for training (default: 64)' '--test-batch-size' 1000 'N' 'input batch size for testing (default: 1000)' '--epochs' 14 'N' 'number of epochs to train (default: 14)' '--lr' 1.0 'LR' 'learning rate (default: 1.0)' '--gamma' 0.7 'M' 'Learning rate step gamma (default: 0.7)' '--no-cuda' 'store_true' False 'disables CUDA training' '--seed' 1 'S' 'random seed (default: 1)' '--log-interval' 10 'N' 'how many batches to wait before logging training status' '--save-model' 'store_true' False 'For Saving the current Model' not and "cuda" if else "cpu" 'num_workers' 1 'pin_memory' True if else '../data' True True 0.1307 0.3081 True '../data' False 0.1307 0.3081 True 1 for in 1 1 if "mnist_cnn.pt" if '__main__' 3. Build the image Build a custom image under the same level of the directory, the target container image in this example is registry.cn-shanghai.aliyuncs.com/tensorflow-samples/mnist:pytorch-1.4-cuda10.1-cudnn7-devel docker build -t \
 registry.cn-shanghai.aliyuncs.com/tensorflow-samples/mnist:pytorch-1.4-cuda10.1-cudnn7-devel . 4. Push the built mirror to the mirror warehouse created in the East China 1 area for users who are in the Greater China area (Alibaba Cloud). You can refer to the registry.cn-shanghai.aliyuncs.com/tensorflow-samples/mnist:pytorch-1.4-cuda10.1-cudnn7-devel basic operation of mirroring. Submit PyTorch training tasks 1. Install arena $ wget http://kubeflow.oss-cn-beijing.aliyuncs.com/arena-installer-0.3.3-332fcde-linux-amd64.tar.gz
$ tar -xvf arena-installer-0.3.3-332fcde-linux-amd64.tar.gz
$ arena-installer/
$ ./install.
$ yum install bash-completion -y
$ >> ~/.bashrc
$ chmod u+x ~/.bashrc cd echo "source <(arena completion bash)" 2. Use arena to submit training tasks, remember to choose selector as dataset=mnist arena submit tf \
             --name=alluxio-pytorch \
             --selector=dataset=mnist \
             --data-dir=/alluxio-fuse/data:/data \
             --gpus=1 \
             --image=registry.cn-shanghai.aliyuncs.com/tensorflow-samples/mnist:pytorch-1.4-cuda10.1-cudnn7-devel \ "python /mnist.py" 3. And view the training log through arena Train Epoch: 12 [49280/60000 (82%)] Loss: 0.021669
Train Epoch: 12 [49920/60000 (83%)] Loss: 0.008180
Train Epoch: 12 [50560/60000 (84%)] Loss: 0.009288
Train Epoch: 12 [51200/60000 (85%)] Loss: 0.035657
Train Epoch: 12 [51840/60000 (86%)] Loss: 0.006190
Train Epoch: 12 [52480/60000 (87%)] Loss: 0.007776
Train Epoch: 12 [53120/60000 (88%)] Loss: 0.001990
Train Epoch: 12 [53760/60000 (90%)] Loss: 0.003609
Train Epoch: 12 [54400/60000 (91%)] Loss: 0.001943
Train Epoch: 12 [55040/60000 (92%)] Loss: 0.078825
Train Epoch: 12 [55680/60000 (93%)] Loss: 0.000925
Train Epoch: 12 [56320/60000 (94%)] Loss: 0.018071
Train Epoch: 12 [56960/60000 (95%)] Loss: 0.031451
Train Epoch: 12 [57600/60000 (96%)] Loss: 0.031353
Train Epoch: 12 [58240/60000 (97%)] Loss: 0.075761
Train Epoch: 12 [58880/60000 (98%)] Loss: 0.003975
Train Epoch: 12 [59520/60000 (99%)] Loss: 0.085389

Test : Average loss: 0.0256, Accuracy: 9921/10000 (99%) # arena logs --tail=20 alluxio-pytorch set Summary Previously, running the PyTorch program required users to modify the PyTorch adapter code to be able to access data in HDFS. Using  Alluxio, we were able to quickly develop and train models without any additional work to modify PyTorch code or manually copy HDFS data. This approach is further simplified by setting up the entire environment within Kubernetes.

Deep Learning at Alibaba Cloud with Alluxio: How To Run PyTorch on HDFS

This article presents the collaboration of Alibaba, Alluxio, and Nanjing University in tackling the problem of Deep Learning model training in the cloud. Various performance bottlenecks are analyzed with detailed optimizations of each component in the architecture. This content was previously published on , featuring Alibaba Cloud Container Service Team's case study (White Paper ). Alluxio's Engineering Blog here Our goal was to reduce the cost and complexity of data access for Deep Learning training in a hybrid environment, which resulted in over 40% reduction in training time and cost. 1.New trends in AI: Kubernetes-Based Deep Learning in the Cloud Background Artificial neural networks are trained with increasingly massive amounts of data, driving innovative solutions to improve data processing. Distributed Deep Learning (DL) model training can take advantage of multiple technologies, such as: Cloud computing for elastic and scalable infrastructure Docker for isolation and agile iteration via containers and Kubernetes for orchestrating the deployment of containers Accelerated computing hardware, such as GPUs The merger of these technologies as a combined solution is emerging as the industry trend for DL training. Data access challenges of the conventional solutions Data is often stored in private data centers rather than in the cloud for various reasons, such as security compliance, data sovereignty, or legacy infrastructure. A conventional solution for DL model training in the cloud typically involves synchronizing data between the private data storage and its cloud counterpart. As the size of data grows, the associated costs for maintaining this synchronization becomes overwhelming: Copying large datasets: Datasets continually grow, so eventually it becomes infeasible to copy the entire dataset into the cloud, even if copying to a distributed high-performance storage such as GlusterFS. Transfer costs: New or updated data needs to be continuously sent to the cloud to be processed. Not only does this incur transfer costs, but it is also costly to maintain because it is often a manually scripted process. Cloud storage costs: Keeping a copy of data in the cloud incurs the storage costs of the duplicated data. A hybrid solution that connects private data centers to cloud platforms is needed to mitigate these unnecessary costs. Because the solution separates the storage and compute architectures, it introduces performance issues since remote data access will be limited by network bandwidth. In order for DL models to be efficiently trained in a hybrid architecture, the speed bump of inefficient data access must be addressed. 2. Container and data orchestration based architecture We designed and implemented a model training architecture based on container and data orchestration technologies as shown below: Core components of system architecture is a popular container orchestration platform, which provides the flexibility to use different machine learning frameworks through containers and the agility to scale as needed. Alibaba Cloud Kubernetes (ACK) is a Kubernetes service provided by Alibaba Cloud. Kubernetes is an open-source Kubernetes-based cloud-native AI platform used to develop, orchestrate, deploy, and run scalable, portable machine learning workloads. Kubeflow supports two TensorFlow frameworks for distributed training, namely the parameter server mode and AllReduce mode. Kubeflow is an open-source data orchestration system for hybrid cloud environments. By adding a layer of data abstraction between the storage system and the compute framework, it provides a unified mounting namespace, hierarchical cache, and multiple data access interfaces. It is able to support efficient data access for large-scale data in various complex environments, including private, public, and hybrid cloud clusters. Alluxio Recently introduced features in Alluxio have further cemented its utility for machine learning frameworks. A POSIX filesystem interface based on FUSE provides an efficient data access method for existing AI training models. , jointly developed by the Alluxio and Alibaba Cloud Container Service teams, greatly simplify the deployment of Alluxio in the Kubernetes ecosystem. Helm charts 3. Training in the Cloud Initial Performance In the performance evaluation using the ResNet-50 model, we found that upgrading from NVidia P100 to NVidia V100 resulted in a 3x improvement in the training speed of a single card. This computing performance improvement puts additional pressure on data storage access, which also poses new performance challenges for Alluxio's I/O. This bottleneck can be quantified by comparing the performance of Alluxio with a synthetic data run of the same training computation. The synthetic run represents the theoretical upper limit of the training performance with no I/O overhead since the data utilized by the training program is self-generated. The following figure measures the image processing rate of the two systems against the number of GPUs utilized. Initially, both systems perform similarly but as the number of GPUs increases, Alluxio noticeably lags behind. At 8 GPUs, Alluxio is processing at 30% of the synthetic data rate. Analysis and Performance Optimization To investigate what factors are affecting performance, we analyzed Alluxio's technology stack as shown below, and identified several major areas of performance issues. For a complete list of optimizations we applied, please refer to the full-length whitepaper. 1. Filesystem RPC overhead Alluxio file operations require multiple RPC interactions to fetch the requested data. As a virtual filesystem, its master processes manage filesystem metadata, keeps track of where data blocks are located, and fetch data from underlying filesystems. Solution: We reduced this overhead by caching metadata and not allowing the cache to expire throughout the duration of the workload by setting to . The cache size and expiration time, determined by and respectively should be set to keep the cached information relevant throughout the entire workload. In addition we extend the heartbeat interval to reduce the frequency of worker updates by setting the property to be 2 minutes or longer. alluxio.user.metadata.cache.enabled true alluxio.user.metadata.cache.max.size alluxio.user.metadata.cache.expiration.time alluxio.user.worker.list.refresh.interval 2. Data caching and eviction strategies As Alluxio reads data from the underlying storage system, the worker will cache data in memory. As the allocated memory fills up, it will choose to evict the least recently accessed data. Both of these operations are asynchronous and can cause a noticeable overhead in data access speeds, especially as the amount of data cached on a node nears saturation. Solution: We configured Alluxio to avoid evicting data and disabled features that would cache duplicate copies of the same data. By default, Alluxio caches data read from the underlying storage system, which is beneficial for access patterns where the same data is requested repeatedly. However, DL training typically reads the entire dataset, which can easily exceed Alluxio’s cache capacity as datasets are generally on the order of terabytes. We can tune how Alluxio caches data by setting to , which will avoid evicting data to reduce the load on each individual worker, and setting to false to avoid caching additional copies. alluxio.user.ufs.block.read.location.policy alluxio.client.block.policy.LocalFirstAvoidEvictionPolicy alluxio.user.file.passive.cache.enabled 3. Alluxio and FUSE configuration for handling numerous concurrent read requests Reading files through the FUSE interface is fairly inefficient using its default configuration. FUSE reads are handled by the libfuse non-blocking thread pool that does not necessarily recycle its threads. The thread pool is configured with max_idle_threads hardcoded to a value of 10. When this value is exceeded, threads will be deleted instead of recycled. Solution: Because libfuse2 does not expose the value of as a configuration value, we patched its code to be able to configure a much larger value to better support more concurrent requests. max_idle_threads 4. Impact of running Alluxio in containers on its thread pool Running Java 8 in a containerized environment may not fully utilize a machine’s CPU resources due to how the size of thread pools is calculated. Solution: Before , thread pools would be initialized with a size of 1, which severely restricts concurrent operations. We can set the number of processors directly by setting the Java flag . patch 191 for Java 8 -XX:ActiveProcessorCount Results After optimizing Alluxio, the single-machine eight-card training performance of ResNet50 is improved by 236.1%. The performance gains are reflected in the four-machine eight-card scenario with a performance loss of 3.29% when compared against the synthetic data measurement. Compared to saving data to an SSD cloud disk in a four-machine eight-card scenario, Alluxio's performance is better by 70.1%. The total training time of the workload takes 65 minutes when using Alluxio on four machines with eight cards each, which is very close to the synthetic data scenario that takes 63 minutes. Compared with training via SSD on the cloud, Alluxio saves 45 minutes in time and 40.9% in costs. 4. Summary and future work In this article, we present the challenges of using Alluxio in a high-performance distributed deep learning model training and dive into our work in optimizing Alluxio. Various optimizations were made to improve the experience of AlluxioFUSE performance in high concurrency reading workload. These improvements enabled us to achieve a near-optimal performance when executing a distributed model training scheme in a four-machine eight-card scenario with ResNet50. For future work, Alluxio is working on enabling page cache support and FUSE layer stability. Alibaba Cloud Container Service team is collaborating with both the Alluxio Open Source Community and Nanjing University through Dr. Haipeng Dai and Dr. Rong Gu. We believe that through the joint effort of both academia and the open-source community, we can gradually reduce the cost and complexity of data access for Deep Learning training when compute and storage are separated to further help advance AI model training on the cloud. 5. Special Thanks Special thanks to Dr. Bin Fan, Lu Qiu, Calvin Jia, and Cheng Chang from the Alluxio team for their substantial contribution to the design and optimization of the entire system. Their work significantly improved the metadata cache system and advanced Alluxio’s potential in AI and Deep Learning. Authors Yang Che, Sr. Technical Expert at Alibaba Cloud. Yang is heavily involved in the development of Kubernetes and container related products, specifically in building machine learning platforms with cloud-native technology. He is the main author and core maintainer of . GPU shared scheduling Rong Gu, Associate Researcher at Nanjing University and core maintainer of the Alluxio Open Source project. Rong received a Ph.D. degree from Nanjing University in 2016 with a focus on big data processing. Prior to that, Rong worked on R&D of big data systems at Microsoft Asia Research Institute, Intel, and Baidu.

Efficient Model Training in the Cloud with Kubernetes, TensorFlow, and Alluxio Open Source

This article describes how can accelerate the training of deep learning models in a hybrid cloud environment when using Intel’s open source platform, powered by oneAPI. Details on the new architecture and workflow, as well as Alluxio’s performance benefits and benchmarks results will be discussed. The original article can be found on . Alluxio Analytics Zoo Alluxio's Engineering Blog 1.  Deep Learning in Hybrid Environments Architecture Evolution to Hybrid Mode Traditionally, data processing and analytics systems were designed, built, and operated with compute and storage services as one monolithic platform, residing in an on-premises data warehouse. While simple to manage and performant, this architecture with deeply coupled storage and compute is often challenging to provide applications elasticity and scale more resources for one type without scaling the other. More users are moving towards a hybrid model, combining resources from both cloud and on-premises environments. This model practices an alternative architecture to leave the data where it resides, typically in the on-premises data warehouse, but launch a separate compute layer as needed. The hybrid model allows compute and storage resources to be scaled independently, leading to numerous advantages: On-premise machines can be fully utilized by storage services because there is no competition for resources from compute services No resource contention: : There are no idle compute resources because clusters are launched on demand in the cloud No compute downtime : Long-running batch jobs or ephemeral ad-hoc queries can share the same set of data without making separate copies No data duplication Challenges to Deliver Fast I/O for Deep Learning Although the hybrid architecture provides flexibility and cost advantages, there are additional challenges for deep learning analytics when training on big data. Deep learning training involves numerous trials of different neural network models and different hyper-parameters using the same set of data. In addition, the size of training datasets has been continuously growing. There is a huge overhead cost in loading all this data for each trial when the training data is stored in a remote storage system. A common practice today to manage data across hybrid environments is to copy data to a storage service residing in the compute cluster before running deep learning jobs. Typically, users use commands like “distCP” to copy data back and forth between the on-premise and cloud environments. While this looks easy, it typically requires a manual process which is slow and error-prone. To address the I/O challenges of training deep learning models in hybrid environments and leverage Intel’s oneAPI performance optimizations, we developed and tested a new architecture/workflow integrating Alluxio in the Analytics Zoo platform, powered by oneAPI. 2. A New Architecture & Workflow with Analytics Zoo and Alluxio What is Analytics Zoo Analytics Zoo is an open source unified analytics and AI platform developed by Intel to seamlessly unite several deep learning applications into an integrated pipeline. Users can transparently scale from running sample jobs on a laptop to processing production scale big data on large clusters. It supports: Writing or inline with Spark code for distributed training and inference TensorFlow PyTorch Native deep learning (TensorFlow/Keras/PyTorch/BigDL) support in Pipelines Spark ML Directly running Ray programs on big data clusters through . RayOnSpark Plain Java/Python APIs for (TensorFlow/PyTorch/BigDL/OpenVINO) Model Inference What is Alluxio Alluxio is an open-source data orchestration layer for data analytics. It provides high performance to data analytics or machine learning systems like Analytics Zoo, serving as a distributed caching layer to prevent reading data repeatedly from remote data sources. Compared to other solutions, Alluxio provides the following advantages in a hybrid cloud environment with “zero-copy burst” capabilities to burst data processing to the cloud: : When a storage system is mounted onto Alluxio, only its metadata is initially loaded. Alluxio only caches data as an application requests for it. This on demand behavior allows burst data processing to the cloud, eliminating the need to manually copy data from an on-premise cluster to the cloud. Compute-driven data on-demand Alluxio intelligently caches data close to applications, replicates hot data, or evicts stale data based on data access patterns. Data Locality: : Alluxio can be elastically scaled along with the analytics frameworks, including container orchestrated environments. Data Elasticity : Alluxio provides data abstraction with different common APIs including the HDFS API, S3 API, POSIX API and others. Existing applications built for analytical and AI workloads can run directly on this data without any changes to the application itself Common APIs for data access Setup and Workflow The following figure is the architecture that integrates Alluxio with Analytics Zoo for fast and efficient deep learning workloads: On-premise or remote data stores are mounted onto Alluxio. Analytics Zoo application launches deep learning training jobs by running Spark jobs, loading data from Alluxio through the distributed file system interface. Initially, Alluxio has not cached any data, so it retrieves it from the mounted data store and serves it to the Analytics Zoo application while keeping a cached copy amongst its workers. This first trial will run at approximately the same speed as if the application was reading directly from the on-premise data source. In subsequent trials, Alluxio will have a cached copy, so data will be served directly from the Alluxio workers, eliminating the remote request to the on-premise data store. Note that the caching process is transparent to the user; there is no manual intervention needed to load the data into Alluxio. However, Alluxio does provide commands like “ ” to preload the working dataset to warm the cache if desired. There is also a “ ” command to reclaim the cache storage space without purging data from underlying data stores. distributedLoad free 3. Benchmark Results This section summarizes Alluxio’s performance testing and benchmark results for the integrated workflow. Environment We run experiments in a 7-node Spark cluster (1 instance as the master node and the remaining as worker nodes) deployed by AWS EMR. The benchmark workload is inception v1 training, using the ImageNet dataset stored in AWS S3 in the same region. As the baseline, the Spark cluster is directly accessing the dataset from the S3 bucket. This is compared to a setup where Alluxio is installed on the Spark cluster, with the S3 bucket mounted as its under filesystem. The following table details the specific environment configurations: Result Comparison We measured data loading performance when running an inception training on ImageNet data by using Analytics Zoo. The measured time includes training data and test data loading time. The average load time with and without Alluxio is 579 and 369 seconds, respectively. This is approximately a 1.5x speedup when Analytics Zoo uses Alluxio for loading the ImageNet training and testing data. Note that, the input data is located in S3 in the same region of the compute. (Source: Alluxio’s internal performance testing) The following figure shows that with Alluxio, variation in performance (15.9 second) is also much lower than the baseline variation (32.3 second). This indicates that Alluxio not only helps the average loading time but also makes performance more consistent. 4. Conclusion By leveraging Alluxio as a data layer on Analytics Zoo, the hybrid cloud solution provides acceleration of data loading in Analytics Zoo applications and deep learning analytics on big data systems. Our Alluxio’s internal performance benchmark testing shows this architecture is approximately a 1.5x speedup when Analytics Zoo uses Alluxio for loading the ImageNet training and testing data. Continued advancements in artificial intelligence applications have brought deep learning to the forefront of a new generation of data analytics development. There is an increasing demand from organizations to apply deep learning technologies to their big data analysis pipelines. On behalf of the entire Alluxio open source community, we encourage our readers to give this solution a try, and invite you to ask questions in our whenever you encounter any issues. community slack channel Special thanks to Intel’s Jennie Wang and Louie Tsai for their valuable Analytics Zoo’s technical consultation & support.

ALLUXIO

alluxio.com

-- employees

Since n.d.

COMPANY RANKING

ALLUXIO

EVERGREEN INDEX #1225

Representing this company? Click here to claim and customize this page!

Alluxio WIKI

Apply to get your Tech Company News Page live on HackerNoon.com today!

ALLUXIO

alluxio.com

-- employees

Since n.d.

COMPANY RANKING

ALLUXIO

EVERGREEN INDEX #1225

RELATED COMPANIESRANK

STORIES ABOUT ALLUXIO

Representing this company? Click here to claim and customize this page!

Alluxio WIKI

Apply to get your Tech Company News Page live on HackerNoon.com today!