It is my opinion that too many programmers jump too soon to scale-out systems before making sure that every node is as highly optimized as it could be.
Data will always be able to outgrow hardware's capability of processing it in a timely manner. Parallel systems are critical to handle database tables with billions of rows; file systems with hundreds of millions of files; and NoSql stores with an ever increasing number of KV pairs or documents. So a distributed system becomes necessary at some point.
The problem comes when the threshold for scaling out is set too low. A process gets too slow and instead of optimizing the code, they immediately try breaking it up and distributing it. So instead of needing a dozen servers to handle a big problem, the algorithms are inefficient enough that it takes 100 servers or more to solve the same problem in a reasonable amount of time.
I am working on a distributed data system https://didgets.com/ that handles all kinds of data. I have focused on making sure that every node can process large amounts of data in an efficient manner. It can be so much faster when you don't have to coordinate between too many pieces and more of the data is located close to the CPU processing it.
I recommend the whitepaper Scalability! But at what COST? A single threaded system with data locality can often run in shorter time than a system that is networked.
I also recommend the the document "latencies every developer should know" updated by Jeff Dean.
I say this as most of the time a computer is waiting for IO from main memory or SSD or network or spinning disks. The more sequential you can get memory into the CPU the better. Row major memory pattern, structures of arrays or arrays of structures depending.
I write multithreaded software so I feel people reach for multiprocessing software due to multithreading being harder to get right than putting load balancer in front of it.
I write and journal of concurrency, threading and distributed systems in my GitHub, mainly ideas4.
I was reading your ideas4 list and came across #18 Data Structure Synchronization. It referenced RocketSet which I had not heard before. A brief introduction to their system told me that they did things very similarly to how I implemented file system tags and DB columns in my Didgets system. My tag objects can also be used to do content indexing and creating 'Dictionaries' that are incredibly fast. The bonus is that for each column, the index and the data are one and the same. When you do CRUD operations, you don't have to update more than one copy of the data.
Wow thank you for reading it. I really appreciate that.
I am enamoured with RockSet's converged indexes. They solve the problems of WHERE queries, columnular analytic workloads and row based iteration of data.
I think we tend to look at data is being relatively static and don't denormalise as much as we could for performance and data locality so we bear with slow Microservices and distributed systems with lots of IO. Once data is in Postgres you don't shift it to other machines that often.
I think the next step of distributed systems is key rebalancing and key SCHEDULING. I plan to design a system that shifts data to create nodes where particular queries are fast with data locality due to that server having all the data needed to fulfil the query. Autoshard at Google is interesting too. It requires denormalisation and data synchronization. I am looking at multimaster postgres or writing my own simple synchronizer. But it is eventually consistent.
I also have implemented multiversion concurrency control solution and an early raft implementation that needs to be added to a server. I tend to build components then plug them together. I am looking at Google's spanner and TrueTime.
I shall look at your solution.
How do you avoid the duplicate copying of data with a single copy of the data? How is it columnular or indexed by value?
Each of my data objects called Didgets (short for Data Widgets) that store tag or column information is a simple Key-Value store. I store each unique value once and have links between it and any keys mapped to it.
So if you have a 1 million row table of U.S. customers that has a 'state' column, then each state value (e.g. 'California', 'Iowa', 'Florida', etc.) is only stored once. Each value is referenced counted so if you had 50,000 customers who lived in California that value would have a reference count of 50,000. There would then be 50,000 links between the 'California' value and their respective keys (in this case the row numbers).
If one of your customers moves from California to Texas, you update that value which just decrements the reference count for California and increments the reference count for Texas. Then the row key for that customer is re-linked or mapped from pointing to the California value to pointing to the Texas value.
A query like "SELECT name, state, zip FROM <table> WHERE state LIKE 'T%';" causes the code to find all values in the state column that start with the letter T (Texas, Tennessee). It will then find all the keys mapped to those two values. Then it will load in the data for the name and zip columns and find any values mapped to those keys.
It is incredibly fast partly because it can be multi-threaded (one thread finds the names mapped to the keys while another one finds the zip codes mapped). Analytics are fast too since all the values are reference counted. It can instantly tell you the top 10 states where your customers live.
What's also coincidental is that I read a Quora post recently of network model databases where references are direct. They have the linking problem which CODASYL worked towards solving. Where you need to update all links when there are changes. My idea 18 data synchronization system is to produce one system to handle it. It also could be used for cache invalidation which is an interesting problem in itself.
Your solution also reminds me of a graph database where nodes and vertices/ edges are explicitly stored but in your case you are using reference counting which is new to me.
Thanks for sharing your knowledge on this.
From a logical point of view, explicit storage of links should be more efficient than a hash join or nested loop join. But is there a tradeof in write amplification? Joins in Postgres are materialised at query time but they're efficient due to btrees, whereas in your model the data links are all materialised. I think links in graph databases such as neo4j and dgraph are materialised as links directly too.
Updating values may or may not result in links being changed. For example if you imported data where 'Illinois' was misspelled as 'Ilinois' for the 10,000 customers who lived there; updating the value to the correct spelling might not change any of the links.
All the links are stored within a hashed set of data blocks where similar links are stored near each other. This helps minimize any write amplification. You might be able to update 10,000 links while only needing to write out a few blocks to disk.
It's helpful to plan for scale out if it will be needed, though. It can be difficult to build that later if it wasn't planned for.
On the other hand, you do need to consider scaling up. You can get a HPE ProLiant DL385 Gen10 Plus with dual 64-core Epycs and 8 TB of ram and almost half a petabyte of flash storage. If you're starting from zero, it's likely a long way until that's not enough, by then bigger servers might be easily obtainable, IBM has a power server that goes to 64 TB, but then you're dealing with IBM and power.
If you do need a distributed system, if it's possible, you want to design things as you said, with as little coordination as possible during processing. Thinking about coordination in the system design early can help you make it easier to separate later. Coordination on a single host isn't as expensive as across hosts, but it's not free either, so it's not wasted work to consider early.
Comments
It is my opinion that too many programmers jump too soon to scale-out systems before making sure that every node is as highly optimized as it could be.
Data will always be able to outgrow hardware's capability of processing it in a timely manner. Parallel systems are critical to handle database tables with billions of rows; file systems with hundreds of millions of files; and NoSql stores with an ever increasing number of KV pairs or documents. So a distributed system becomes necessary at some point.
The problem comes when the threshold for scaling out is set too low. A process gets too slow and instead of optimizing the code, they immediately try breaking it up and distributing it. So instead of needing a dozen servers to handle a big problem, the algorithms are inefficient enough that it takes 100 servers or more to solve the same problem in a reasonable amount of time.
I am working on a distributed data system https://didgets.com/ that handles all kinds of data. I have focused on making sure that every node can process large amounts of data in an efficient manner. It can be so much faster when you don't have to coordinate between too many pieces and more of the data is located close to the CPU processing it.
I recommend the whitepaper Scalability! But at what COST? A single threaded system with data locality can often run in shorter time than a system that is networked.
I also recommend the the document "latencies every developer should know" updated by Jeff Dean.
I say this as most of the time a computer is waiting for IO from main memory or SSD or network or spinning disks. The more sequential you can get memory into the CPU the better. Row major memory pattern, structures of arrays or arrays of structures depending.
I write multithreaded software so I feel people reach for multiprocessing software due to multithreading being harder to get right than putting load balancer in front of it.
I write and journal of concurrency, threading and distributed systems in my GitHub, mainly ideas4.
I was reading your ideas4 list and came across #18 Data Structure Synchronization. It referenced RocketSet which I had not heard before. A brief introduction to their system told me that they did things very similarly to how I implemented file system tags and DB columns in my Didgets system. My tag objects can also be used to do content indexing and creating 'Dictionaries' that are incredibly fast. The bonus is that for each column, the index and the data are one and the same. When you do CRUD operations, you don't have to update more than one copy of the data.
Wow thank you for reading it. I really appreciate that.
I am enamoured with RockSet's converged indexes. They solve the problems of WHERE queries, columnular analytic workloads and row based iteration of data.
I think we tend to look at data is being relatively static and don't denormalise as much as we could for performance and data locality so we bear with slow Microservices and distributed systems with lots of IO. Once data is in Postgres you don't shift it to other machines that often.
I think the next step of distributed systems is key rebalancing and key SCHEDULING. I plan to design a system that shifts data to create nodes where particular queries are fast with data locality due to that server having all the data needed to fulfil the query. Autoshard at Google is interesting too. It requires denormalisation and data synchronization. I am looking at multimaster postgres or writing my own simple synchronizer. But it is eventually consistent.
I also have implemented multiversion concurrency control solution and an early raft implementation that needs to be added to a server. I tend to build components then plug them together. I am looking at Google's spanner and TrueTime.
I shall look at your solution.
How do you avoid the duplicate copying of data with a single copy of the data? How is it columnular or indexed by value?
Each of my data objects called Didgets (short for Data Widgets) that store tag or column information is a simple Key-Value store. I store each unique value once and have links between it and any keys mapped to it.
So if you have a 1 million row table of U.S. customers that has a 'state' column, then each state value (e.g. 'California', 'Iowa', 'Florida', etc.) is only stored once. Each value is referenced counted so if you had 50,000 customers who lived in California that value would have a reference count of 50,000. There would then be 50,000 links between the 'California' value and their respective keys (in this case the row numbers).
If one of your customers moves from California to Texas, you update that value which just decrements the reference count for California and increments the reference count for Texas. Then the row key for that customer is re-linked or mapped from pointing to the California value to pointing to the Texas value.
A query like "SELECT name, state, zip FROM <table> WHERE state LIKE 'T%';" causes the code to find all values in the state column that start with the letter T (Texas, Tennessee). It will then find all the keys mapped to those two values. Then it will load in the data for the name and zip columns and find any values mapped to those keys.
It is incredibly fast partly because it can be multi-threaded (one thread finds the names mapped to the keys while another one finds the zip codes mapped). Analytics are fast too since all the values are reference counted. It can instantly tell you the top 10 states where your customers live.
Here is a short video showing how fast it can do it compared to the same data set stored in Postgres. https://www.youtube.com/watch?v=OVICKCkWMZE
That's really interesting.
What's also coincidental is that I read a Quora post recently of network model databases where references are direct. They have the linking problem which CODASYL worked towards solving. Where you need to update all links when there are changes. My idea 18 data synchronization system is to produce one system to handle it. It also could be used for cache invalidation which is an interesting problem in itself.
Your solution also reminds me of a graph database where nodes and vertices/ edges are explicitly stored but in your case you are using reference counting which is new to me.
Thanks for sharing your knowledge on this.
From a logical point of view, explicit storage of links should be more efficient than a hash join or nested loop join. But is there a tradeof in write amplification? Joins in Postgres are materialised at query time but they're efficient due to btrees, whereas in your model the data links are all materialised. I think links in graph databases such as neo4j and dgraph are materialised as links directly too.
Updating values may or may not result in links being changed. For example if you imported data where 'Illinois' was misspelled as 'Ilinois' for the 10,000 customers who lived there; updating the value to the correct spelling might not change any of the links.
All the links are stored within a hashed set of data blocks where similar links are stored near each other. This helps minimize any write amplification. You might be able to update 10,000 links while only needing to write out a few blocks to disk.
Do you have a link to the whitepaper you referenced?
Nevermind. I found it once I figured out the full title was "Scalability! But at what cost?"
http://www.frankmcsherry.org/assets/COST.pdf
https://gist.github.com/jboner/2841832
It's helpful to plan for scale out if it will be needed, though. It can be difficult to build that later if it wasn't planned for.
On the other hand, you do need to consider scaling up. You can get a HPE ProLiant DL385 Gen10 Plus with dual 64-core Epycs and 8 TB of ram and almost half a petabyte of flash storage. If you're starting from zero, it's likely a long way until that's not enough, by then bigger servers might be easily obtainable, IBM has a power server that goes to 64 TB, but then you're dealing with IBM and power.
If you do need a distributed system, if it's possible, you want to design things as you said, with as little coordination as possible during processing. Thinking about coordination in the system design early can help you make it easier to separate later. Coordination on a single host isn't as expensive as across hosts, but it's not free either, so it's not wasted work to consider early.