RDBMS and Apache Geode Data Movement: Low Latency ETL Pipeline By Using Cloud-Native Event Driven Microservices

Editor's Notes

• #3: [ Heather ] Special thanks to Anu, from HCSC, and Jeff, from Pivotal Data. They were instrumental in getting this working in production, and they originally told this story at Cloud Foundry Summit this year.
• #4: [ Paul ] HCSC is the largest customer-owned health insurer in the US and 4th largest overall. We operate as BCBS in 5 states across the country, our largest being Illinois and Texas. We are 15 million members strong, supported by 2100 IT employees. We are a purpose driven organization whose focus is to stand with our members in sickness and in health.
• #5: [ Heather ] 15 million members -> they expect data to be consistent and fast Hospitals and providers Legacy systems not cutting it -> inconsistent and slow We learned that with Geode we would be able to store the data in an in memory data grid and make it available to our consumers really quickly. Getting massive data in many sources -> membership, claims, benefits 1 million claims processed per day So how do we move data at this scale from so many places into Geode? And how do we continue to keep our data always consistent with the source systems as the data changes? We have 15 million members who are trying to get their information via our websites and apps. They expect their data to be consistent across platforms and they expect it to be available fast. It’s not just members, though; hospitals and providers also need to be able to get their information from us quickly and consistently. The legacy systems we had in place to serve up all of this data just weren’t cutting it anymore. The raw data coming out of the systems wasn’t always consistent and wasn’t available nearly quick enough. We learned that with Geode we would be able to store the data in an in memory data grid and make it available to our consumers really quickly. We uncovered a new challenge: getting a massive amount of data into Geode. This was complicated because we have many different source systems: membership, claims, benefits, with huge amounts of data. Think, 1 million claims processed per day. So how do we move data at this scale from so many places into Geode? And how do we continue to keep our data always consistent with the source systems as the data changes?
• #6: [ Heather ] Simple EXTRACT TRANSFORM LOAD (ETL) process source needs to move to destination typical: file extract of source db, run through etl process, inserted into target Runs in batch ETL drag and drop tools Business rules become complicated with complex data (drag and drop at large scale!?) Not horizontally scalable/parallelizable. Not real time since it runs as batch ------------------------- Let’s start with a simple EXTRACT TRANSFORM LOAD (ETL) process. We have a source system that needs to be moved to a destination. A typical ETL process could be a file extract of the source database, which is then run through the ETL process and inserted into the destination. Traditionally, ETL processes run in batch or with ETL tools such as “drag and drop” where you map the source to the destination. Business rules and transformations can become complicated, especially when you need to move complex data. Unfortunately, it’s not horizontally scalable because it’s not easily parallelizable. It also isn’t real time since it runs as a batch process, which means we’ll inevitably have stale data.
• #7: But creating an event driven process helps us achieve near real time data transfer. [ Paul ] As the data changes in the source system it can write events to an event table and your ETL process can be triggered based on these events. The ETL process reads the event table, which contains the key information we can use to identify the changed data and how to process it. Then it reads from the source database and extract the information It then can transform it to the destination format, and load to Geode. But, we’ll still see some problems with this architecture when there are large loads and high volume. This process is also not as horizontally scalable as we would like. We will potentially hit a bottleneck reading from the event table in the case of hundreds or millions of events happening in one day. You can try to handle this read in batch, but you’ll still need to loop through and process them sequentially. So we need to multithread the darker box and scale it up. But we can only scale so much in a single JVM, maybe 1 or 2 hundred threads. It still will not give you the throughput you need to get through millions of records.
• #8: Restructuring our monolithic architecture into microservices allows us to scale where we need it. [ Heather ] Looked at bottleneck -> pieces doing I/O (event reader and load process) Separate event-reading into microservice One responsibility: read events and push out to another microservice Next microservice uses the event data to extract from the source db Transform the raw data into the format of the data in Geode The loader takes that transformed data and aggregates it and does batch PUTs into Geode. Capability to scale the E/T microservice and process events in parallel Throughput we need because we can scale where necessary By removing the monolith, we’ve gained some flexibility in scaling, but we’ve introduced complexity around communicating between microservices Messages need to be passed between each component responsible for boilerplate code and configurations Takes our focus away from business rules. --------------------------------------- We started looking at which parts of the process were a bottleneck and identified those as the pieces doing I/O, the listener and the load process. We separated out event-reading into its own microservice and gave it only one responsibility: read the events and push it out into another microservice. The next microservice has the job of extracting and transforming the data into the destination format. And finally, the loader aggregates them all together and does a batch put into Geode. This gives us the capability to scale the extract and transform microservice vertically as well as horizontally and process the events in parallel. Now we can get the throughput we need because we can scale where necessary. By removing the monolith, we’ve gained some flexibility in scaling, but we’ve introduced complexity around communicating between microservices Messages need to be passed between each component, and that makes us responsible for boilerplate code and configurations that take our focus away from business rules. // Breaks apart the monolith. Not sure what the arrows represent, http calls?
• #9: So we introduced Spring Cloud Stream to manage message-driven communication between our microservices. [ Paul ] A quick introduction to Spring Cloud Stream; it is a framework for building message-driven microservices. It abstracted the boilerplate code and configurations away with its out of the box configuration for message middleware (like RabbitMQ or Kafka). SCS provides three binding interfaces: a source, processor, and sink. The source defines an interface which allows messages or events to enter the processing stream. From there processors or sinks can subscribe to the channel to receive messages. Processors take an an input channel and publish messages or events to an output channel. Sinks end the processing flow. They subscribe to input channels but do not provide any outputs.
• #10: [ Paul Still ] Our listener became a source because its only purpose was to read events and publish them to the stream. The extract/transform service made sense as a processor because it takes input and transforms it, publishing the transformed information through the stream. Our loader is a sink because it collected the transformed information and wrote it into Geode, ending the stream.
• #11: “Let’s jump into a demo” Our code is available on github if you would like to follow along. Start with the read-me, it has all the steps necessary to get PCF Dev setup as well as Apache Geode. We be starting from the “Running the demo” section in the readme. [ Heather ] STATE: Apps deployed to PCF dev, and gemfire started, schemas created in mysql. PCF Dev – show apps. We have PCF Dev running locally. We’ve already deployed our three spring boot apps – the source, processor, and sink Services - RabbitMQ has also already been created. We’re simple using that as our message middleware. SCS is configured to bind to rabbit on application start. MySQL - We are using MySQL as our source database. We’ve created our tables – Customer, Item, Order_Item, Customer_Order, and Db_Events Source_Key is the key of the record that was inserted, updated, or deleted The Source_Group column allows us to use one event table to record events that occur across many tables. We don’t need to have a customer_event table, an item_event table, etc. Action_Code tells us whether this was an insert update or delete Status_Code is code that identifies the event as unprocessed, processed, or failed The Event_ts column records time that the action happened in the source system. That way, if a single record has many updates happen in a short amount of time, we can use this timestamp to find the newest change. Jump to gfsh, we’ve created our locator, server, and regions `list regions`. We have a customer, item, and customerOrder region. 2_create_initial_data – run entire file 3_synthetic_event – create all events Jump to gfsh, run all 4 queries Triggers – walk through one, crud it up Insert data bit by bit, switching to gemfire to show Code it up - paul
• #12: [ Heather ] SCs allowed us to simplify our data flow without having to change the way we think about each core component Handled communication between services We now have the flexibility to add and remove sources, processors, sinks anywhere in the flow, at a very low cost So what happens when we do have more than one data source? Example of data coming in from a file SCS allowed us to simplify our data flow without having to change the way we think about each core component. It handled communication between our services, while allowing us the flexibility to add and remove sources, processors, or sinks, anywhere in the flow, at a very low cost. All component based with independent services providing their own value. What happens we have multiple data sources, how do we bring it all together. Spring cloud stream is a flow. If we need to add additional processing steps or sources we can do that easily. For example one of our source data came in from a file. Which was easy to add to this current flow. When we read from the file we created a source and processor specific to the file and the file format. From there we changed the initial processors to transform to an standard enterprise format in which the second processor and sink could handle. This built that abstraction layer removing the coupling to the source system format If there are any changes or if the additional source system can provide real time data it is just a change to that specific source/processor and the rest of the system stays the same. // Is there a way we can tie this back to our problem slide? WE called out different source systems providing different pieces of the member experience (Add two more … and click through them “and another, and another”)
• #13: [ Paul ] Spring cloud stream has error handling out of the box. SCS will move any failed messages to an error queue, we just needed to turn it on! After we started looking at the messages that failed, we found that if we had just let them reprocess, they probably would have processed successfully on the second or third time. To reduce operations we wrote a sink process which read the error queue and in the event table mark the row as unprocessed. This allows the source to re-read the message and try again. At this time we added a column to keep track of the number of times this event was processed. After a few attempts we change the status code to a different value so operations can manually look at the messages and take action.
• #14: [ Heather ] Didn’t want to add triggers Don’t touch legacy Poor performance Change Data Capture tool scans database change logs and writes events Initial load – millions of records Didn’t want a new process Wrote synthetic events Kicked off entire bulk loading process With all the data loaded, and events being written by the CDC tool, we have an extremely fast and consistent data store that can aggregate data from any number of source systems. This strategy has given our consuming applications the confidence to utilize our services over the legacy systems they’ve been using for years. -------------------------------------------- In our case, we needed to pull data from a legacy system that we wanted to avoid touching. Our source system was a legacy system and we really didn't want to touch it to create the event triggers. To solve this we used an external Change Data Capture (CDC) tool to monitor the database change logs to write to the event table. As we mentioned before, when we go to production the first time, we already have millions of records in our source database, but nothing in Geode. We didn't want to write a new process to move the initial data to our target database. All we had to do was scan our table get our key information and insert it into the event table as synthetic events. This kicked off the entire process to move the data in bulk.
• #15: [ Paul ]
• #16: [ Heather ] Thank you for your time everyone! If you missed out on any of HCSC’s other sessions this week, we told an epic story about becoming Cloud Native with the help of Spring and Geode so please check them out online.