Unlocking Operational Intelligence from the Data Lake: Part 3 - The Operational Data Lake in Action

Mat Keep
May 27, 2016
#Technical

Welcome back to our 3-part blog series on creating operational data lakes:

In part 1, we discussed the rise of the data lake, the role of Hadoop, and the challenges in integrating the data lake with operational applications
In part 2, we covered the critical capabilities you need to evaluate in an operational database for your data lake, and a recommended technology design pattern for integrating the database with the data lake
In this final installment, we’ll wrap up by covering real world examples and best practices from industry leaders

If you want to get a head-start and learn about all of these topics now, just go ahead and read the Operational Data Lake white paper.

Best Practice Case Studies

The following examples demonstrate how leading companies are using the design pattern discussed in part 2, and shown again below, to operationalize their data lakes.

Figure 1: Design pattern for operationalizing the data lake

UK’s Leading Price Comparison Site

Out-Innovating Competitors with MongoDB, Hadoop, Microservices, Docker, and the Cloud

The UK’s leading price comparison provider, and one of the country’s best known household brands has standardized on MongoDB Enterprise Advanced as the default operational database across its microservices architecture. The company’s online comparison systems need to collect customer details efficiently and then securely submit them to a number of different providers. Once the insurers' systems respond, the company can aggregate and display prices for consumers. At the same time, MongoDB generates real-time analytics to personalize the customer experience across the company’s web and mobile properties.

With the previous generation of systems, all application state was stored in the database, and then imported every 24 hours from backups into the company’s data warehouse. But that approach presented several critical issues:

No real-time insight as the analytics processes were working against aged data.
Application changes broke the ETL pipeline.
The management overhead increased as more applications were added and data volumes grew.

As the company transitioned to microservices, the data warehousing and analytics stacks were also modernized. While each microservice uses its own MongoDB database, the company needs to maintain synchronization between services, so every application event is written to a Kafka queue. Event processing runs against the queue to identify relevant events that can then trigger specific actions – for example customizing customer questions, firing off emails, presenting new offers and more. Relevant events are written to MongoDB, enabling the user experience to be personalized in real time as customers interact with the service.

All events are also written into Hadoop where they can be aggregated and processed with historical activity, in conjunction with additional customer data from the insurance providers. This enables the company to build enriched data views such as user profiles or policy offers. The models are then imported into the operational MongoDB databases to further enhance user experience, and maximize cross and upsell opportunities.

As a result of its modernized architecture, the company has established a leading position in the highly competitive price comparison market, while achieving 2x faster time to market after migrating from its former SQL Server relational database to MongoDB, and enabled continuous delivery to push new features live every day.

Leading Global Airline

Revenue Optimization with MongoDB, Spark, and Hadoop

Through a series of mergers and acquisitions, the airline’s customer data was scattered across 100 different systems. As a result, the company had no way to gain a single, 360 degree view of the business in order to analyze customer behavior, identify gaps in product portfolios, or present a consistent and personalized passenger experience across airline brands.

With its data lake built on Hadoop, the airline initially evaluated Apache HBase to serve operational applications, but found the column-oriented data model to be restrictive. The need to pre-define column families meant that any functional change in the online applications would break HBase’s single view schema. The lack of secondary indexes prevented the database from efficiently handling the array of queries needed for customer care applications and real-time analytics.

After further technology evaluation, the company has been able to bring together customer profiles into a single view stored in MongoDB, distributed across multiple data centers to service the online web, mobile and call center applications. All customer interactions, ticket sales and account data are processed and stored in MongoDB, and then written to the company’s Hadoop cluster where Spark machine learning jobs are run to build customer classifications, optimize ticket pricing and identify churn risks. These are then retrieved by MongoDB to serve the online applications. Spark processes are also run against the live operational data in MongoDB to update customer classifications and personalize offers in real time, as the customer is live on the web or speaking with the call center.

With MongoDB, Hadoop, and Spark powering its modern data architecture, the airline is meeting its goals of delivering personalized experiences to the millions of passengers it carries every year, while optimizing ticket prices and enhancing service offerings that reduce competitive threat.

Stratio

Integrates Apache Spark and MongoDB to Unlock New Customer Insights for One of the World's Largest Banks

The Stratio Apache Spark-certified Big Data (BD) platform is used by an impressive client list including BBVA, Just Eat, Santander, SAP, Sony, and Telefonica. The company has implemented a unified real-time monitoring platform for a multinational banking group operating in 31 countries with 51 million clients all over the world. The bank wanted to ensure a high quality of service and personalized experience across its online channels, and needed to continuously monitor client activity to check service response times and identify potential issues. The application was built on a modern technology foundation including:

Apache Flume to aggregate log data Apache Spark to process log events in real time MongoDB to persist log data, processed events and Key Performance Indicators (KPIs).

The aggregated KPIs, stored by MongoDB enable the bank to analyze client and systems behavior in real time in order to improve the customer experience. Collecting raw log data allows the bank to immediately rebuild user sessions if a service fails, with analysis generated by MongoDB and Spark providing complete traceability to quickly identify the root cause of any issue.

The project required a database that provided always-on availability, high performance, and linear scalability. In addition, a fully dynamic schema was needed to support high volumes of rapidly changing semi-structured and unstructured JSON data being ingested from a variety of logs, clickstreams, and social networks. After evaluating the project’s requirements, Stratio concluded MongoDB was the best fit. With MongoDB’s query projections and secondary indexes, analytic processes run by the Stratio BD platform avoid the need to scan the entire data set, which is not the case with more simple datastores.

Working with some of the world’s largest enterprises, Stratio has seen data lakes growing in use, with MongoDB’s distributed design and dynamic schema a great fit as it is impossible to predict what type of data structures need to managed at scale.

Learn more by reading the Stratio case study.

Conclusion

Hadoop-based data lakes are enabling organizations to efficiently capture and analyze unprecedented volumes of data generated from connected devices and users. But without being able to expose that data to operational applications, users are struggling to maximize returns on their Hadoop investments. The longer it takes to surface insight to operational processes, the less valuable that insight is. With its flexible data model, powerful in-database analytics, distributed, scale-out architecture, and low latency performance, MongoDB provides the best solution to operationalize the data lake.

Learn more by reading the Operational Data Lake white paper. Unlocking Operational Intelligence from the Data Lake

About the Author - Mat Keep

Mat is director of product and market analysis at MongoDB. He is responsible for building the vision, positioning and content for MongoDB’s products and services, including the analysis of market trends and customer requirements. Prior to MongoDB, Mat was director of product management at Oracle Corp. with responsibility for the MySQL database in web, telecoms, cloud and big data workloads. This followed a series of sales, business development and analyst / programmer positions with both technology vendors and end-user companies.

← Previous

Multiplexing Golang Channels to Maximize Throughput

The Go language is great for concurrency, but when you have to do work that is naturally serial, must you forgo those benefits? We faced this question while rewriting our database backup utility, mongodump , and utilized a “divide-and-multiplex” method to marry a high-throughput concurrent workload with a serial output. The need for concurrency In MongoDB, data is organized into collections of documents . When reading from a collection, requests are often preempted, when other processes obtain a write lock in that collection. To prevent stalls from reducing overall throughput, you can enqueue reads from multiple collections at once. Thus, a previous version of mongodump concurrently read data across collections, to achieve maximum throughput. However, since the old mongodump wrote each collection to a separate file, it did not work for two very common use cases for database backup utilities: 1) streaming the backup over a network, and 2) streaming the backup directly into another instance as part of a load operation. Our new version was designed to support these use cases. To do that, while preserving the throughput-maximizing properties of concurrent reads, we leveraged some Golang constructs -- including reflection and channels -- to safely permit multiple goroutines to concurrently feed data into the archive. Let me show you how. The archive format The new mongodump archive consists of two sections: the prelude and the body. The prelude records the number of collections backed up, and other metadata. In the body, we interleave slices of data from the collections. Each collection's last slice is a special "EOF" slice. Let’s get a bird’s eye view of the archive layout: Because this format supports interleaving slices of collection data, we can concurrently back up multiple databases and collections to a single file, allowing users to easily compress archives, or pipe mongodump through ssh to mongorestore on another server . But how does this layout bridge the serial/concurrent chasm? Writing to an archive In our new mongodump , we implement concurrent reading with a fixed, user-specified number of goroutines. Suppose our MongoDB server has two databases, db1 and db2 , that we want to back up. Let's tell mongodump to use two goroutines, by passing -j2 : mongodump -j2 --archive=backup.archive --host mongod.example.com The diagram shows how mongodump allocates goroutines. Each goroutine gets pegged to a particular collection to read data from ( collection1, collection2 , etc above) and serves as an input source to the multiplexer. These two goroutines are used as arguments to Golang’s Select function. Here’s its function signature: func Select(cases []SelectCase) (chosen int, recv Value, recvOK bool) Each SelectCase maps to one of the multiplexer’s input sources. recv holds a BSON document received from an input source. recvOK is true until all data from that input source has been completely received, then a recvOK of false signals completion. To tie this all together, we introduce a special control SelectCase to signal one of two states. The first state is when the goroutine associated with a particular collection begins working on a new collection. The second state is when the all input sources have been completely read. Select is the core of our design to bridge the chasm between the archive's serial nature and mongodump ’s concurrency. Here’s how the multiplexer is created: func NewMultiplexer(out io.WriteCloser) *Multiplexer { controlInput := make(chan *MuxIn) multiplexer := &Multiplexer { control: controlInput, selectCases: []reflect.SelectCase { reflect.SelectCase { Dir: reflect.SelectRecv, Chan: reflect.ValueOf(controlInput), Send: reflect.Value{}, }, } return multiplexer } And here’s how we use it: // Run multiplexer until it receives an EOF on the control channel. func (mux *Multiplexer) Run() { for { index, value, recv := reflect.Select(mux.selectCases) EOF := !recv // note that the control channel is always at index 0 if index == 0 { if EOF { return } muxInput, _ := value.Interface().(*muxInputSource) mux.selectCases = append(mux.selectCases, reflect.SelectCase{ Dir: reflect.SelectRecv, Chan: reflect.ValueOf(muxInput.collection), Send: reflect.Value{}, }) } else { if EOF { mux.writeEOF() mux.selectCases = append(mux.selectCases[:index], mux.selectCases[index+1:]...) } else { document, _ := value.Interface().([]byte) mux.writeDocument(document)[] } } } } The main goroutine creates the multiplexer and calls go Run() on it. In our example, we have two goroutines each reading data from a distinct collection. The Select function blocks until at least one of the goroutines is ready to send data. Then, we read the reflect value (a BSON document) off that goroutine’s channel and write the document to the archive file (lines 22-23). This process continues for each collection until it is completely read. But what do we do when we’ve finished backing up data from one collection, or finished backing up all the data? This is where the control channel comes in. First, we consider how to reassign a goroutine once it has completely read all data from a collection. When a goroutine begins reading data from a new collection, it signals this to the multiplexer by sending a signal on the control channel. This causes the multiplexer to register the new collection in the select cases it chooses from (lines 12-16). Once the goroutine for that collection has completely read all the data, it sends an EOF which removes it from the multiplexer’s select cases (line 20). Then the goroutine reading that collection gets reassigned to the next collection, so all goroutines are kept busy until all collections have been backed up. (The Select function makes a copy of all the select cases before selecting, so we can safely update the select cases that comprise the multiplexer's inputs while the multiplexer is running.) How does the multiplexer know when to stop? After the main mongodump process invokes go Run() , it waits for the backup goroutines to completely back up all the collections. Once these goroutines finish, the main mongodump process closes the control channel, and multiplexer goroutine knows it's done. Reading from an archive Reading from a mongodump archive file is straightforward. Once we’ve parsed the prelude, we know all collections to expect during the restore operation. In the body, we simply read each sliceStart we encounter, and assign a goroutine to insert all documents for that collection into the appropriate collection. Each available goroutine is assigned to a unique collection so that, the same as backing up, we restore collections concurrently. By the way -- that curious magic number at the beginning of the archive -- 0x8199e26d (-2120621459) ? The first byte, 6d , is the hex representation of the ASCII character ’m’. The remaining three bytes represent the earth symbol, ’♁’. Together, they form ’m♁’. So it's an allusion to how well suited MongoDB is to handling humongous data. But there's another reason we chose it. The first four bytes of a BSON document is the document's length as a signed integer. The magic number was specifically chosen because it's invalid as the start of a BSON document (since it’s negative). The tool we designed to back up MongoDB collections to a single archive file is straightforward and easy to understand. Using a tidy file format and Golang's powerful concurrency, our new backup and restore tools are faster, simpler, and more featureful than before the rewrite.

May 26, 2016

Next →

How Edenlab Built a High-Load, Low-Code FHIR Server to Deliver Healthcare for 40 Million Plus Patients

The Kodjin FHIR server has speed and scale in its DNA. Edenlab, the Ukrainian company behind Kodjin , built our original FHIR solution to digitize and service the entire Ukrainian national health system. The learnings and technologies from that project informed our development of the Kodjin FHIR server. At Edenlab, we have always been driven by our passion for building solutions that excel in speed and scale. With Kodjin, we have embraced a modern tech stack to deliver unparalleled performance that can handle the demands of large-scale healthcare systems, providing efficient data management and seamless interoperability. Eugene Yesakov, Solution Architect, Author of Kodjin Built for speed and scale While most healthcare projects involve handling large volumes of data, including patient records, medical images, and sensor data, the Kodjin FHIR server is based on a system developed to handle tens of millions of patient records and thousands of requests per second, to ensure timely access and efficient decision-making for a population of over 40 million people. And all of this information had to be processed and exchanged in real-time or near real-time, without delays or bottlenecks. This article will explore some of the architectural decisions the Edenlab team took when building Kodjin, specifically the role MongoDB played in enhancing performance and ensuring scalability. We will examine the benefits of leveraging MongoDB's scalability, flexibility, and robust querying capabilities, as well as its ability to handle the increasing velocity and volume of healthcare data without compromising performance. About Kodjin FHIR server Kodjin is an ONC-certified and HIPAA-compliant FHIR Server that offers hassle-free healthcare data management. It has been designed to meet the growing demands of healthcare projects, allowing for the efficient handling of increasing data volumes and concurrent requests. Its architecture, built on a horizontally scalable microservices approach, utilizes cutting-edge technologies such as the Rust programming language, MongoDB, ElasticSearch, Kafka, and Kubernetes. These technologies enable Kodjin to provide users with a low-code approach while harnessing the full potential of the FHIR specification. A deeper dive into the architecture approach - the role of MongoDB in Kodjin When deciding on the technology stack for the Kodjin FHIR Server, the Edenlab team knew that a document database would be required to serve as a transactional data store. In an FHIR Server, a transactional data store ensures that data operations occur in an atomic and consistent manner, allowing for the integrity and reliability of the data. Document databases are well-suited for this purpose as they provide a flexible schema and allow for storing complex data structures, such as those found in FHIR data. FHIR resources are represented in a hierarchical structure and can be quite intricate, with nested elements and relationships. Document databases, like MongoDB, excel at handling such complex and hierarchical data structures, making them an ideal choice for storing FHIR data. In addition to supporting document storage, the Edenlab team needed the chosen database to provide transactional capabilities for FHIR data operations. FHIR transactions, which encompass a set of related data operations that should either succeed or fail as a whole, are essential for maintaining data consistency and integrity. They can also be used to roll back changes if any part of the transaction fails. MongoDB provides support for multi-document transactions , enabling atomic operations across multiple documents within a single transaction. This aligns well with the transactional requirements of FHIR data and ensures data consistency in Kodjin. Implementation of GridFS as a storage for the terminologies in Terminology service Terminology service plays a vital role in FHIR projects, requiring a reliable and efficient storage solution for terminologies used. Kodjin employs GridFS , a file system within MongoDB designed for storing large files, which makes it ideal to handle terminologies. GridFS offers a convenient way to store and manage terminology files, ensuring easy accessibility and seamless integration within the FHIR ecosystem. By utilizing MongoDB's GridFS, Kodjin ensures efficient storage and retrieval of terminologies, enhancing the overall functionality of the terminology service. Kodjin FHIR server performance To evaluate the efficiency and responsiveness of the Kodjin FHIR server in various scenarios we conducted multiple performance tests using Locust, an open-source load testing tool. One of the performance metrics measured was the retrieval of resources by their unique ids using the GET by ID operation. Kodjin with MongoDB achieved a performance of 1721.8 requests per second (RPS) for this operation. This indicates that the server can efficiently retrieve specific resources, enabling quick access to desired data. The search operation, which involves querying ElasticSearch to obtain the ids of the searched resources and retrieving them from MongoDB, exhibited a performance of 1896.4 RPS. This highlights the effectiveness of polyglot persistence in Kodjin, leveraging ElasticSearch for fast and efficient search queries and MongoDB for resource retrieval. The system demonstrated its ability to process search queries and retrieve relevant results promptly. In terms of resource creation, Kodjin with MongoDB showed a performance of 1405.6 RPS for POST resource operations. This signifies that the system can effectively handle numerous resource-creation requests. The efficient processing and insertion of new resources into the MongoDB database ensure seamless data persistence and scalability. Overall, the performance tests confirm that Kodjin with MongoDB delivers efficient and responsive performance across various FHIR operations. The high RPS values obtained demonstrate the system's capability to handle significant workloads and provide timely access to resources through GET by ID, search, and POST operations. Conclusion Kodjin leverages a modern tech stack including Rust, Kafka, and Kubernetes to deliver the highest levels of performance. At the heart of Kodjin is MongoDB, which serves as a transactional data store. MongoDB's capabilities, such as multi-document transactions and flexible schema, ensure the integrity and consistency of FHIR data operations. The utilization of GridFS within MongoDB ensures efficient storage and retrieval of terminologies, optimizing the functionality of the Terminology service. To experience the power and potential of the Kodjin FHIR server firsthand, we invite you to contact the Edenlab team for a demo. For more information On MongoDB’s work in healthcare, and to understand why the world’s largest healthcare companies trust MongoDB, read our whitepaper on radical interoperability .

May 31, 2023