Source-linked AI summary
Data Mining Using High Performance Data Clouds: Experimental Studies Using Sector and Sphere
Robert L Grossman, Yunhong Gu
TL;DR
Large distributed datasets require infrastructure that can store and process data across high-performance wide-area networks while limiting the cost of distribution. The paper designs Sector/Sphere, combines persistent storage with parallel stream processing, and reports faster wide-area benchmark performance than Hadoop, alongside distributed network-analysis applications.
Problem
Existing distributed mining systems often move data to shared processors, while Hadoop and related systems are designed primarily for tightly coupled data-center racks rather than loosely coupled wide-area environments.
Method
The paper designs Sector as a distributed indexed-file storage cloud and Sphere as a compute cloud that applies user-defined functions to Sector-managed data streams.
Results
Sector/Sphere was approximately 2.4–2.6 times faster than Hadoop on Terasort and about 1.6–1.9 times faster on Terasplit in wide-area experiments.
Takeaways & Limitations
Sector/Sphere processed large datasets distributed across the continental United States with an approximately 80% performance penalty relative to placing all data on one rack.
Takeaways & Limitations
The reported Hadoop comparison may be affected by tuning differences, including Hadoop’s better performance on 1 Gb/s than 10 Gb/s NICs and limited tuning for 10 Gb/s networks.
Abstract
from arXiv · showhide
We describe the design and implementation of a high performance cloud that we have used to archive, analyze and mine large distributed data sets. By a cloud, we mean an infrastructure that provides resources and/or services over the Internet. A storage cloud provides storage services, while a compute cloud provides compute services. We describe the design of the Sector storage cloud and how it provides the storage services required by the Sphere compute cloud. We also describe the programming paradigm supported by the Sphere compute cloud. Sector and Sphere are designed for analyzing large data sets using computer clusters connected with wide area high performance networks (for example, 10+ Gb/s). We describe a distributed data mining application that we have developed using Sector and Sphere. Finally, we describe some experimental studies comparing Sector/Sphere to Hadoop.
1. INTRODUCTION
Sector/Sphere is a distributed high-performance data-mining system that combines persistent distributed storage with parallel stream processing. It targets large datasets across clusters connected by wide-area high-performance networks and is evaluated through applications and Hadoop comparisons.
- System design: Sector provides persistent, indexed-file storage for large datasets distributed across nodes.It replicates data for longevity, lower retrieval latency, and parallel processing.
- System design: Sphere applies the same user-defined function independently to each record and data segment managed by Sector.This creates natural parallelism and often processes data in place without moving it.
- System design: Sector/Sphere uses specialized network protocols to scale across wide-area high-performance networks.The architecture combines distributed indexed files, uniform stream processing, and network-aware scaling.
- Evaluation and applications: The paper presents the system design, a distributed network-data-mining application, experiments, and comparisons with Hadoop using Terasort and Terasplit.The application searches for emergent behavior in distributed network data.
- Paper scope: The paper organizes these contributions across background, Sphere, Sector, networking, experiments, an application, and summary sections.It describes a later version of the system in more detail than earlier work.
2. BACKGROUND AND RELATED WORK
The paper contrasts conventional data-movement-heavy cluster mining with Sector/Sphere’s persistent, in-place processing model. Sector is designed for loosely coupled systems and wide-area 10 Gb/s networks, while Hadoop and related file systems target tightly coupled data-center racks.
- Motivation: Conventional cluster and grid mining often moves data to shared processors, spending substantial time transporting it.Message passing and grid services commonly scatter data, compute, gather results, and repeat the process.
- Motivation: Sector/Sphere instead stores data persistently and processes it in place when possible.This model lets data wait for the task or query rather than requiring repeated transfers.
- Network design: Sector is designed for wide-area high-performance 10 Gb/s networks and uses specialized protocols such as UDT to use their bandwidth.The paper identifies relatively small inter-cluster bandwidth as an assumption in earlier storage-cloud work.
- Storage architecture: Sector targets loosely coupled distributed systems with peer-to-peer management, unlike GFS and HDFS, which use more tightly coupled centralized architectures.Sector also assumes data is divided into files, while GFS and HDFS divide it into blocks.
- Programming models: MapReduce extracts, shuffles, and reduces data, whereas Sphere supports arbitrary user-defined processing over Sector-managed streams.MapReduce is described as a special case using common Map and Reduce functions.
3. DESIGN OF SPHERE
Sphere provides a stream-processing model for applying user-defined functions to data managed by Sector. It divides datasets into independently processed segments, assigns them to processing elements, and routes outputs according to the stream definition.
- Overview: A Sphere operator applies a user-defined function to every record in a Sector-managed dataset.The command abstracts disk I/O and the loop needed to process all records.
- Overview: Sphere is a stream programming model in which user-defined functions transform input streams into output streams.The model differs from MapReduce’s restriction to key-value data and Map and Reduce functions.
- Data organization: Datasets are divided into multiple files, which can be processed in parallel along with their replicas.The SDSS example uses 64 files of approximately 15.6 GB each.
- Computing model: Sphere processes data segments through Sphere Processing Elements, or SPEs, which read segments, apply operators, buffer results, and write outputs.Segments may be records, groups of records, or entire files.
- Output handling: Sphere can return results to the originating Sector node, write them locally, or shuffle them to specified nodes.This output placement is determined by the definition of the output stream.
- Scheduling: Sphere assigns roughly balanced segments subject to user-defined minimum and maximum sizes, preferring SPEs on the same machine.Segments from the same file are not processed simultaneously unless avoiding that would leave an SPE idle.
4. DESIGN OF SECTOR
Sector provides persistent, indexed storage for large distributed datasets, with replication, access control, and networking features designed for high-performance wide area environments.
- Storage services: Sector provides persistent storage for large distributed datasets and manages the data required by Sphere operations.It relies partly on local native file systems rather than functioning as a file system itself.
- Access control: Sector controls write access through access control while allowing public or community reading unless additional restrictions apply.Clients must satisfy the relevant server access-control requirements to upload data.
- Replication: Sector replicates files to support archival safety, reduce retrieval latency, and enable parallel processing.Replica counts are checked daily, and additional replicas may be created at randomly selected locations.
- Network support: Sector supports wide area high-performance networks through specialized transport, routing, and messaging protocols with separate interfaces.Its requirements include long-term archival storage, wide area bandwidth utilization, multiple routing and network protocols, and community-oriented access.
- Demonstration: 1.3TB compressed SDSS BESTDR5 data was transported at approximately 8.1 Gb/s over a 10 Gb/s wide area network using six commodity servers.The example demonstrates Sector’s use for archiving and distributing SDSS data to astronomers worldwide.
- Indexed data: Sector divides datasets into files and records, using co-located index files to support random record access.Without an index, Sphere can process data only at the file level and the user must provide parsing and extraction logic.
5. DESIGN OF NETWORKING LAYER
The networking layer supports geographically distributed clusters and large datasets through configurable routing and specialized high-performance transport and messaging protocols.
- Routing: The experimental Sector configuration uses loose management across geographically distributed clusters connected by a high-performance wide area network.A peer-to-peer Chord routing protocol allows nodes to be added and removed easily.
- Future scope: Future Sector routing protocols are intended for both uniform clouds and clouds with widely varying bandwidth and round-trip times.The planned support distinguishes clouds with approximately equal RTTs from non-uniform clouds where bandwidth and RTT vary substantially across clusters.
- Transport: UDT provides rate-based transport for large data flows, sharing bandwidth fairly and backing off when congestion occurs.The protocol is designed for wide area high-performance networks and remains friendly to concurrent TCP flows.
- Messaging: Sector uses the Group Messaging Protocol for message passing between geographically distributed nodes.GMP is a specialized network transport protocol developed for Sector.
6. EXPERIMENTAL STUDIES
The experiments compare Sector/Sphere with Hadoop on Terasort and Terasplit in local and geographically distributed settings. Sector/Sphere is faster in both benchmark settings and shows performance impacts when operating across wide-area network delays, though some comparisons have important qualifications.
- Experimental setup: The experiments use Terasort and Terasplit to compare Sphere with Hadoop on an 8-node local cluster and six geographically distributed servers.The wide-area servers span Chicago, Pasadena, and Greenbelt over 10 Gb/s networks; Hadoop version 0.16.0 is used.
- Benchmarks: Terasplit computes a single entropy-based tree split from sorted data, providing a benchmark for typical data-mining computation.The reported version reads potentially distributed data into a single client to compute the split.
- Qualification: The Terasort comparison is not exact because Sector/Sphere uses one of four available cores while Hadoop uses all four.The authors also attribute some local-area differences to tuning, particularly Hadoop's limited tuning for 10 Gb/s NICs.
- Wide-area results: Sector/Sphere improves performance over Hadoop by approximately 2.4–2.6 on wide-area Terasort, 1.6–1.8 on Terasplit, and 2.1-2.3 on the combined benchmark.These figures are reported as performance improvements measured by the respective benchmarks.
- Local-area results: Sphere is about 1.6–2.3 times faster than Hadoop on Terasort and about 1.2-1.5 times faster on Terasplit in local-area experiments.Both systems scale well on the reported benchmarks, according to the table.
- Wide-area scaling: Across geographically distributed settings, performance impact is approximately 41% for four nodes across two locations and approximately 82% across three locations for Sector/Sphere.The corresponding Hadoop impacts are approximately 41% and 67%, respectively, with RTTs of 16 ms, 55 ms, and 71 ms in the three-location case.
7. SPHERE APPLICATIONS
The Angle application uses Sector and Sphere to process anonymized, geographically distributed TCP packet data and detect emergent behavior through temporal clustering. Its experiments examine distributed-file workloads and cluster-center changes across time windows.
- Angle: Angle collects anonymized IP data at geographically distributed sensors, stores packet windows in Sector, and uses Sphere to identify anomalous or suspicious TCP behavior.Sensors zero packet content, hash source and destination IPs, package moving windows as pcap files, and transfer them to associated Sector nodes.
- Angle: Angle processes approximately 575 pcap files daily, totaling approximately 7.6GB and 97 million packets, across four installed sensor locations.Approximately 300,000 pcap files had been collected to date.
- Angle: Sphere aggregates pcap files by source IP or another specified entity, computes feature files, and processes them to identify emergent behavior.The pipeline first anonymizes and manages packet data through Sector before Sphere aggregates and computes features.
- Emergent-cluster detection: Sphere forms clusters for successive temporal windows and flags emergent clusters when cluster behavior changes significantly after preceding windows remain relatively stable.The window length d determines the temporal aggregation used for cluster comparison.
- Emergent-cluster detection: The statistic δj measures cluster-center movement between adjacent windows; its graph is described as quite choppy for ten-minute windows.The paper also presents the statistic for one-day windows, where emergent clusters support scoring functions.
- Scoring: The scoring function combines emergent-cluster centers and variances, with constants λk summing to 1, and scores feature vectors against the emergent clusters.The maximum is taken over emergent clusters k.
- Experimental workload: Table 3 evaluates Sphere clustering as the workload grows from 1 to 300,000 distributed pcap files, corresponding to approximately 500 to 100,000,000 feature vectors.Feature files are aggregated and clustered after each pcap file produces one feature file.
8. SUMMARY AND CONCLUSION
The paper presents Sector/Sphere as a cloud infrastructure for mining large distributed datasets over high-performance wide-area networks. It reports large-scale processing, network-traffic analysis, and faster benchmark performance than Hadoop.
- Summary and conclusion: Sector/Sphere is a cloud-based infrastructure for data mining large distributed datasets across clusters connected by high-performance wide-area networks.The infrastructure is open source and has supported several distributed data mining applications.
- Summary and conclusion: The infrastructure combines Sector’s storage cloud with Sphere’s compute cloud and uses a specialized networking layer for distributed processing.The reported performance penalty across continental U.S. locations is approximately 80% relative to data located on a single rack.
- Summary and conclusion: Sector/Sphere computes clusters for the Angle network-traffic application on over 300,000 distributed files.Angle detects emergent behavior in network traffic.
- Summary and conclusion: On a wide-area testbed, Sector/Sphere was approximately 2.4–2.6 times faster than Hadoop on Terasort and about 1.6–1.9 times faster on Terasplit.Terasplit computes a single split in a classification and regression tree.