PostgreSQL on Hadoop
CC Attribution - NonCommercial - ShareAlike 3.0 Unported:
You are free to use, adapt and copy, distribute and transmit the work or content in adapted or unchanged form for any legal and non-commercial purpose as long as the work is attributed to the author in the manner specified by the author or licensor and the work or content is shared also in adapted form only under the conditions of this
Steinbach, Carl
Bridging the Divide with Distributed Foreign Tables Apache Hadoop is an open-source framework that enables the construction of distributed, data-intensive applications running on clusters of commodity hardware. Building on a foundation initially composed of the MapReduce programming model and Hadoop Distributed Filesystem, in recent years Hadoop has expanded to include applications for data warehousing (Apache Hive), ETL (Apache Pig), and NoSQL column stores (Apache HBase). In this talk we describe recent work done at Citus Data that makes it possible to run a distributed version of PostgreSQL on top of Hadoop in a manner that combines the rich feature set and low-latency responsiveness of PostgreSQL with the scalability and performance characteristics of Hadoop. This talk will begin with a high level overview of Hadoop that focuses on its distributed storage layer and block-based replication model. Next we will look at the data model of the Apache Hive data warehousing system and explain how it enables features such as schema-on-read, support for semi-structured data, and pluggable storage formats. Finally, we will describe how we leveraged these ideas and Foreign Data Wrappers to build a distributed version of PostgreSQL. This version runs natively on Hadoop clusters and seamlessly integrates with other components in the Hadoop ecosystem.
 My name is Carl Steinbach. I'm an engineer at Citus Data. I'm also a member of the PMC and a committer on the Apache Hive project, which is kind of like a database project related to Hadoop. And in the past, I've worked at a bunch of enterprise software companies, most recently, Cloud Era, where I was team lead for the Hive team. I've also spent time at Informatica, NetApp, where I worked on storage security products, and Oracle, where I started my professional career working on the database server. My other claim to fame is that I wrote that blog post about Tom Lane's email habits, which I think a lot of you probably read. Thank you. Thank you very much. So hopefully, this presentation lives up to that blog post. Anyway, I've given talks on this topic a couple times in the past. And from those experiences, I've learned that it's a good idea to start upfront by making sure that we're all on the same page about what a distributed analytic database is. I think probably 95% of people, when you say relational database, they automatically think OLTP, the standard sort of create, update, delete, use case scenario with transactions. And an analytic database is definitely not OLTP. It's typically referred to as an OLAP database. So the kind of things that you're going to do with an OLAP database include processing, let's say, clickstream events, or processing log files, or processing some kind of event or fact type data. And typically, people are going to do aggregations on those data or big joins. But in terms of the way that the actual database behaves, one thing that sort of defines OLAP use cases is large sequential scans, where typically you're looking at the entire table as part of a query. Another thing is they're not really transaction oriented. So a lot of the stuff that you see in OLTP databases don't really apply. And then I think finally, and this is probably most important to the rest of this presentation, in terms of the performance characteristics of OLAP databases, they are typically IO bound. So if you're looking for sort of the performance bottleneck in the system, it's not going to be the CPU. It's typically going to be the rate at which the database process can actually pull information off of disk. So a lot of work over the past 10 or 20 years has gone into like figuring out how to reduce the impact of that bottleneck or eliminate it altogether. So one thing that also relates to what we're going to discuss is the enterprise storage model. And this is something that's evolved over, let's say, the past 20 or 30 years. And the goal of this is to basically solve the problem of availability and accessibility for data in the context of, let's say, a large enterprise, a company with 10, 100, 1,000 employees, and terabytes, and possibly even petabytes of data. So obviously, that data is not going to sit on one computer. It's going to sit on a bunch of computers. And you want somehow to make it accessible to a large number of people in your organization. So you need to be able to access it over a network of some sort. And the sort of solution that's evolved to this over time and really that a whole industry has been built around is basically structuring your data center such that you have a pool of storage. And this could be in the form of a NAS or a SAN. And then it's connected via some kind of switching network fabric to a pool of compute resources. So there are a couple problems with this. One problem that customers feel immediately is cost. This is a really expensive solution. This is a lot more expensive than going out and buying a hard drive at Fry's Electronics or Best Buy or something like that. I mean, you're talking about spending upwards of $1,000 per terabyte. Probably even more conventionally. And then add to that the fact that these systems have sort of a built-in scalability problem. And the scalability problem is that really big pipe that sits between your island of compute resources and your island of storage resources. So as you attempt to scale out your storage pool, you're going to have to make that pipe bigger and bigger and bigger. And in practice, I mean, there are sort of limits to how big you can make that pipe. There are also limits to how consequently how much you can scale the underlying storage tier. And what happens as a consequence is that people tend to sort of partition these resources, which means they'll end up with multiple islands of storage pools and multiple islands of compute workers. And then that adds, on top of it, an additional manageability problem. And there's sort of a whole industry that's grown around just solving that, helping you to shuttle data from one storage pool to another to meet SLAs or jobs that you have running in the background. I also want to say that this diagram is sort of a gross oversimplification. I'm trying to get the main points home. But I wanted to point out that besides just having a storage pool, there are actually, in a sense, multiple different tiers of storage pool with different price performance characteristics. So oftentimes, in the compute pool, you might have an OLAP database, and that database will have its own sort of private storage pool, which will often be, in terms of cost per terabyte, much more expensive than sort of your larger pool of online storage. And consequently, you fall into this usage pattern where you're always sort of moving data between these two pools. Because you can't fit all of the stuff that you want to look at in the sort of fast, small pool of storage. And as a result, there are pretty big management headaches. So over 10 years ago at this point, the folks at Google were building this gigantic search engine. And in order to build the inverted index that they needed to power that search engine, they were collecting these gigantic data sets by doing web crawls. And I think they realized pretty early on that if they got hooked on sort of the NAS or SAN solution, that that would not be just a bottleneck for their data center, but it would really be a bottleneck for their business. So they knew that they had to find a different solution to that problem. And they knew that whatever solution they ended up with, it had to both be inexpensive. And it had to be very scalable. So the sort of design priorities that they set out were, first off, we're going to base this on commodity hardware. We want the hardware that the system is built on top of to be as inexpensive as possible. And sort of a consequence of that is when you go with inexpensive commodity hardware, you expect the stuff to fail all the time. And if you're deploying 1,000 of these boxes, you expect probably one box to fail every day, if not multiple boxes. So you have to have a system that can be resilient in the face of these constant failures and outages. So another really interesting thing about this is that the solution that they came up with, of course, is referred to as the Google File System. But in many ways, it's not really a conventional file system. It's more of, let's say, a distributed block store. And there's an interview online that you can find with Sean Quinlan, who was one of the original architects of this system. He sort of talks about the genesis of how they came up with this sort of set of priorities and the things that they decided were not so important. And I think one of the really critical things is that they weren't worried about supporting or building a POSIX file system. They weren't worried about supporting random reads or random writes. So there's a lot of stuff, a lot of file system dogma that they just dispensed with very quickly. And in the interview with Quinlan, he makes the point that since the people who were building GFS were actually the same people who were building the system on top of GFS that were consuming it, that they were able to be very fluid in terms of defining what the requirements were for this system. And I think that's kind of, it's an interesting thing to think about in terms of how we build systems in the future. That sometimes just these sort of little political disagreements between groups at different layers in the system can really stymie advances like this. So the system that they ended up with was referred to as the Google File System. And this is sort of a high level overview of how it works. So when a file is written to GFS, it's basically split up into a number of fixed size chunks. So in the original GFS implementation, these chunks were 64 megabytes in size, which is a lot larger than the blocks that you find in a traditional file system. But it makes sense in the context of storing files that on average are going to be gigabytes, if not potentially even terabytes in size. And the advantage, of course, of having big blocks, or at least one really significant advantage, is that it reduces the metadata load on whatever system you're using to store the file system metadata. There's less of it to store since the number of objects that it needs to track and keep track of are limited in number. So another sort of significant design element here is that once that file is split up into chunks, those chunks are then parceled out more or less randomly to a set of chunk servers. And three copies, or whatever the replication factor for the file system are, by default, three copies would then be distributed across the chunk servers in the system. So there are, in one sense, this is very different from the way that traditional distributed file systems achieve reliability by using something like RAID or a parity-based striping mechanism, which is more efficient in terms of the amount of disk space that's used. I mean, you're able to, let's say, if you use RAID 5, you're basically trading off a fifth of your storage space for that additional reliability. So this is a lot less efficient in terms of disk space, but because they're actually creating multiple copies of every chunk, they do get some other benefits in return. So one of them is that the load can be distributed very evenly across these chunk servers when the same file is being read by multiple clients. So basically, a coordinator has the option of saying, if these three clients are reading the same file, they don't necessarily have to all talk to the same chunk server. Another advantage of this is that in the system that they devised, they expected not just disk failures, but they also expected to have chunk server failures. They expected the whole node to just disappear and go down. And at that point, if you're basically building chunk servers on top of RAID, the RAID wouldn't matter, right? Because you lose the chunk server as an entire unit. So this way, when a chunk server goes down, you have two other chunk servers available where you can read the same block from. And finally, the other nice characteristic of this is that when a chunk server goes down, since all of the blocks that are on that one chunk server have replicas distributed in a more or less uniform fashion over the remaining chunk servers, the load that's required to read those chunks back and copy them to a new chunk server in order to maintain the replication factor is then distributed evenly across the other servers in the pool. So earlier, I mentioned that the problem that Google was contending with was twofold. They were worried about the cost of the traditional NAS and SAN solution, and they were also worried about scalability. So I think if you look at just the stuff that's described in the original GFS paper, and you hold it against these goals, based on what they articulated in that paper, it's pretty clear that they solved the cost problem. I mean, they articulated a way of building a very scalable file system using commodity components. But I don't think that if you look at just the GFS paper, that they really solve the scalability problem. Because fundamentally, it's still a client server type of file system, where the clients are all going to pull data across the network. And that network is then going to become the IO bottleneck. And it wasn't until a year later when they published the MapReduce paper that the missing puzzle piece appeared. And it was in the MapReduce paper where they said, no, actually, here's how GFS works. This is not a conventional file system where you have clients are pulling data over to the client in order to do work on the data there. Instead, we are going to take the work and push it over to the data. So MapReduce is basically a distributed computing framework that allows you to send work over to the nodes where the data actually exists. And by maintaining data locality, you almost completely eliminate this network bottleneck problem. MapReduce does require some network traffic between nodes. It has a shuffle and sort phase and stuff like that. But at least at a first order, you've basically eliminated this really significant bottleneck. And that, I think, is the key benefit of Hadoop and of HDFS. And I think it's really interesting that if you just look at the GFS paper, that this missing piece is not mentioned anywhere there. But if you go back and you read that paper, you can tell that that's what they're thinking. It's kind of sort of an interesting historical thing. So these papers were published in 2003 and 2004. And at around that same time, Doug Cutting and Mike Cafferella were working on an open source project called Nutch, which was an outgrowth of Apache Lucene. And their goal was to build an open source search engine. So they saw these papers and they realized immediately, hey, this is clearly the way that we have to build Nutch. We need to build a distributed file system equivalent to GFS. And we need to devise our own map reduce implementation. And that's what became Hadoop. So Hadoop is these two things. HDFS, which is Hadoop's version of GFS, and then their own map reduce implementation. So that project came into existence in 2005. It was initially used quite heavily at Yahoo and then Facebook. And in the last, I would say, three to four years, we've seen sort of more general adoption within enterprise data centers. So my previous employer, Cloudera, has sort of built a business around helping companies that are, let's say, already using Oracle or already using SQL Server or something like that to figure out ways of sort of transitioning to this lower cost and more scalable infrastructure. But the fact that people are actually using this in industry has given us an opportunity to get a sense for what are the practical benefits of Hadoop and also what are the drawbacks. So in terms of benefits, I think at the very top of the list is the way that HDFS has commoditized enterprise storage. And this has had really significant effects on in terms of what enterprise organizations are doing with their data and also the amount of data that they're actually retaining on site. So in the past and actually still today, you find that most large organizations have some data that they maintain in an online fashion in their data center. But a lot of the data that they collect is getting written to tape and put on a truck. And the truck drives to a cave somewhere. And it's deposited there. And no one ever looks at it again. And some of you guys might think I'm joking about the cave thing, but you should look up Iron Mountain, which is one of these companies. They literally have a cave. So that's one major thing. Like companies that were formerly putting a lot of data on tape never to be seen again are now maintaining it in HDFS instead. And they're able to derive a lot of benefit from that data. Another big advantage is the way that HDFS allows companies to scale out in a genuinely fashion instead of having to invest in more expensive hardware to scale up. It also gives them the ability to run a system in a fault-tolerant manner, I think in a manner that wasn't really available before with other existing NAS and SAN solutions. And it's also introduced a new level of flexibility in terms of the data that these organizations can operate on. So one of the things about Hadoop is that it is, or HDFS in particular, it's a write-once-read-many-times file system. It's like an online archive file system. So the ability to operate on data in place in the format in which it lands on disk is very attractive. And Hadoop offers flexible tools that allow you basically to extract information from a variety of different formats. So at the same time, I think people are aware that Hadoop has a bunch of drawbacks. And chief among these is MapReduce itself. MapReduce is very powerful in the sense that it's extremely flexible, but a lot of that flexibility comes from the fact that it's so generic or so general and also so low level. So to do even sort of simple operations, you're going to end up writing pages of Java code. And once you've written this code, it's more or less impossible to maintain. I mean, you're going to have trouble remembering a month later what the code you wrote does and your ability to share it with a coworker is close to zero. So part of this relates to the fact that MapReduce doesn't have any schemas. And schemas are really valuable. I think many times when people think about schemas in the context of a relational database, they maybe think that in some ways schemas are the enemy. Because oftentimes you'll find that maybe you have some data that you really want to analyze in the database. But before you can even get it into the database, you have to talk to your DBA. And your DBA is going to spend a couple of days trying to figure out the right schema for this data. So there's a lot of organizational latency built into this. And the reason for that is that if you get the schema wrong, that's an expensive mistake to make. So schemas in the traditional context are a high stakes game. But they also provide a lot of value. Because they allow you to view data in a logical fashion as opposed to having to worry about the underlying physical structure of it. And Hadoop, without schemas, forces you to constantly worry about the underlying physical structure. And then the other missing features I think can quickly be summarized by saying that people really want Hadoop to look more like a database. They want Hadoop, they like the scalability, they like the flexibility, but at the same time, databases look the way they do for a reason. They've evolved over the past 20 or 30 years in response to the needs of users. And they provide a bunch of features that make a lot of sense when what you're trying to do is analyze data. So these include things that help you write code or help you write better code, like optimizers, indexes, and views. And there's also this large ecosystem of tools that run on top of databases that know how to speak things like ODBC and JDBC, and which allow you to do really interesting things with your data, like business intelligence tools, ETL tools for transforming and manipulating your data. And then things like SQL IDEs or other environments for doing data analysis. So the way that the Hadoop community responded to these shortcomings of the sort of core platform was by building a variety of domain-specific languages on top of Hadoop. So the idea was you no longer have to code your job in MapReduce, instead you can use this other language which we compile down to MapReduce. So there were a variety of languages that evolved like this. Google, of course, has something called SAASOL, which they published a paper on. They also have a SQL DSL referred to as Tensing. Within the Hadoop world, we have things like PIG. We also have Cascading, which is kind of like a Java data flow API. The DSL that I think has gained the most interest and adoption within the enterprise software community is Hive. So Hive made the very pragmatic decision instead of inventing our own language, we're going to stick with SQL. Because SQL has some rough edges. It has some bugs. But at least we've learned what those bugs are over the last 30 years, and they're easy to work around. And also, it turns out the entire world basically speaks SQL, including all of those ecosystem tools that I mentioned earlier. So SQL is like the Esperanto of data processing, except that people actually speak it, unlike Esperanto. So anyway, that's the situation we're in. So one thing about Hive that I really want to try to convey is that most people, when you mention it, they think, OK, Hive is like a SQL to map reduce compilation and execution engine, which is very true. That's what it is. But it's also these two other things that I don't think it gets enough credit for being. So one of those things basically is the data model that it provides. And by data model, I mean the way that it basically allows you to map logical structure in the form of schemas to underlying physical data that can be in basically any format you want. So in order to provide that functionality, Hive has two components. One is the Hive Metastore, which basically allows you to think of it as Hive's table catalog. So the Metastore maintains mappings of the form table basically to underlying file in HDFS, or to HDFS location where the data is stored. And it also maintains mappings of table to SIRT name, where SIRTs are basically these plug-in libraries that allow you to handle different formats. So with these two bits of information, when Hive executes a query plan, which is in the form of basically a graph of map reduce jobs, each one of these map reduce jobs gets sent to a data node, which you could think of as HDFS's version of chunk servers in GFS. And the map reduce job, which is expressed in terms of Hive operators, is executed there. But the data, as it's read off of HDFS, is filtered through one of these SIRTs. So an example would be, let's say, a CSV SIRT, or a JSON SIRT, or perhaps an RC file SIRT, which is Hive's columnar format. But in this way, it allows you to do something really cool, which I think people have started to refer to as schema on read as opposed to schema on write. So this is the really cool benefit of Hive, and the thing that I expect will be Hive's lasting impact on the database world, not SQL to map reduce, but this very flexible way of mapping logical structure onto underlying physical data. Yep? So basically, a SIRT, by the way, SIRT stands for serialized, deserialized. So SIRT, S-E-R-D-E-F. So when you are deserializing, you are basically reading from a file, and what you are pulling out of it are rows split into columns with types. And in the reverse direction, you feed in basically these structures that are typed columns, and then it writes it out to the underlying file. So you were saying, could you have a SIRT that goes from CSV to, let's say, JSON, or something like that? You'd actually use two SIRTs for that process. You would use a JSON SIRT and a CSV SIRT, and what you would do would be to construct two tables, one of which is based on top of JSON, the other on top of CSV. And then one interesting thing about Hive is that it's added a variety of sort of extensions to base SQL, including semantics that allow you to easily do things like stream data from one table into another table. So I hope that answers your question. So I think Hive solved a bunch of problems, but in many ways it's also been sort of a victim of its own success. Because as soon as you start making something look like a database, people are like, hey, this is great. I like that it looks like a database, but I wish you would cover that remaining 50% and make it look completely like a database. Where are this missing SQL feature that I'm used to using? Where is the authentication and authorization system? Here's a data type that I expected you to support. It's part of the SQL standard, but I don't see it here. And then another problem with Hive from the standpoint of many users is caused by the fact that it's basically delegating execution to MapReduce. So MapReduce is a very powerful tool, but it makes some compromises which are probably not the best compromises to make in the context of a database engine like Hive. And the big compromise that it makes is that it's batch oriented. So it guarantees basically really high throughput. But it has this built in latency as a result of the fact that when you start a job, you have to talk to the job tracker. And the job tracker has to go and parcel out jobs to each one of the data nodes so that as a result, even if you are running a query against just a small sliver of data, let's say a megabyte or 100 megabytes or something like that. The minimum turnaround time on that query is going to be 15 to 30 seconds. And if you ran the same query, let's say, on Postgres SQL on a single node, you'd expect it to complete in under a second. So with Hive, this whole way of interacting with a database where you do very small exploratory queries to try to gain an understanding of the layout of your data and what's actually there. It's not really feasible or practical with Hive. So in the past year to maybe two years, I think it's fair to say a new generation of distributed analytic databases has emerged, which are largely being designed to solve this set of problems. And in particular, the problem of the latency overhead and inefficiencies that are introduced by using MapReduce. So examples of these systems would include things like Impala from Cloudera, the Apache Drill Project, Google's Dremel as an example of this, also within Google, and CitusDB is another example of that. So I'm going to talk a little bit about CitusDB. But before I do, I wanted to show the common architectural difference that sets these new databases apart from, let's say, the previous generation of MPP databases that emerged over the previous decade. So this looks a lot like what we've seen recently where people have been adapting MPP databases to run on top of Hadoop. And what they're effectively doing is creating a connector between the worker nodes in the MPP cluster and the data nodes in HDFS. And the problem with this approach is that you have the IO bottleneck between these two layers. So you're not achieving data locality for your jobs, and a system like this is just not going to scale. But the solution is, at least if we want to illustrate it graphically, it's pretty straightforward. You get rid of the dedicated MPP data nodes, and you take the processes, the worker processes that were running on those nodes, and you co-locate them on each one of the data nodes. So in effect, what you're doing then is pushing the work down to the data as opposed to pulling the data over to the work. And this allows systems like this to scale out in the same way that you're able to scale out with something like MapReduce. So I wanted to go into some more detail about how this actually plays out in a system like CitusDB. So CitusDB is built on top of PostgreSQL. And if we look at the components in the system, on the left-hand side, we have PostgreSQL clients, ODBC and JDBC driver. And I think this is one of the really significant points I wanted to make, that as far as a client is concerned, as far as a user is concerned, CitusDB looks almost identical to PostgreSQL. So you basically continue to use PSQL or the different tools that you're familiar with. The support for SQL is virtually the same. All of the same system tables are there and stuff like that. But in terms of changes we've made, on the master node, once again, it's more or less stocked PostgreSQL with the addition of our distributed query compiler and execution engine. And then on each one of the HDFS data nodes, we co-locate what is in effect a stock copy of PostgreSQL, along with a foreign data wrapper that we've written that allows us to basically read from HDFS. So I wanted to go into a little bit more detail about how this actually works by giving an example of how query is executed. So to start off, when you create a table on the CitusDB master, you would basically specify a directory or a file in HDFS that that logical table name needs to map to. And then as part of a prior offline step, the master node syncs file system metadata off of the HDFS name node. So this basically allows the master node to map an HDFS file name to a set of blocks and data nodes in the underlying storage pool. And then, once the table is created and once the metadata has been synced, the master node talks to each one of the worker nodes, and it creates a foreign table per block on each one of the data nodes using the HDFS foreign data wrapper. So then subsequently, let's say the user submits a query, a simple aggregation query to find the average salary of managers in the organization. So as a first step, we take that global query and we rewrite it as a set of queries or fragment queries that can be executed on the worker nodes. And those queries then, or there's effectively one query per foreign table where each foreign table maps to a block on each one of those data nodes. The worker nodes execute those fragment queries, generate partial result sets, which are then returned to the master node, which merges those results together and returns a final answer to the client. So there are some interesting details that I wanted to cover at a slightly lower level. So one of them has to do with block awareness. So I had that previous slide where I showed that, in a sense, what we're doing is taking a traditional MPP database architecture. And the most significant change we're making is that we're co-locating the worker processes on the data nodes. But I don't want to leave you with the impression that that's all you have to do in order to achieve data locality. Because the problem is that if that alone is what you do, those MPP worker processes running on the data nodes are still going to have to basically fetch data from all of their neighbors in order to satisfy any query. So it's very important that the master node, when it is scheduling a query, when it is scheduling those fragment queries, that it sends them to the correct data nodes so that those data nodes can do those reads locally. And that's what we refer to as block awareness. So another thing that's worth mentioning is fault tolerance. So our solution to fault tolerance, I think, can quickly be summarized by saying that we're basically leveraging the best features of the two platforms that we're building on top of, Hadoop and PostgreSQL. So what happens if the master node fails? Well, we take advantage of PostgreSQL streaming replication to make sure that the table metadata is copied over to a hot standby. And that solves the problem at the master node level. And at the worker nodes, since we are actually delegating the storage problem as well as the fault tolerance problem to HDFS, if one of those nodes fails, we already know that the actual data has been replicated to a bunch of neighbors and can quickly fail over to them. And the missing piece, the piece that we actually had to implement on our own, is the ability to, if a process or if a query is in the process of executing, detect that a node which is responsible for some of these fragment queries has gone down and to reassign those queries dynamically to other replicas in the system. So one of the things that I think, yep? It's the last of the replicas. Right. Then you're screwed, basically. Yeah. You would get an error back. That's true. And you would get an error back quickly because it would determine that it can't communicate with any of the three replicas in that situation. So there is a limit to, yeah, no, fair question. So back when I was first talking to the guys at Citus Data about joining the company, they were talking about what they had done with CitusDB and how they had built it on top of PostgreSQL. I wasn't very familiar with PostgreSQL at the time. I was mainly familiar with things like MySQL and in particular with things like Hive. And as they were telling me about some of the different features of PostgreSQL that they were leveraging, they mentioned foreign data wrappers. And I got really excited about this because foreign data wrappers, I quickly realized, are really a much more powerful version of Hive-SERDs. And using foreign data wrappers, we are able to achieve the same novel benefits that separate Hive from sort of traditional databases. The ability to do schema on read as opposed to schema on write, which introduces this whole new level of flexibility. And very significantly, schema on read is more than about just flexibility. It's also, in a sense, about performance. Because if you can read data in place, regardless of the format, without first having to go through this process of loading it, in other words, translating it or converting it into the native format of the database before you can read it, that's a significant performance in terms of overall latency. And just as a quick side note, a couple weeks ago or maybe a month ago, someone posted a performance comparison between Hive and Amazon Redshift. And this was carried on Hacker News. And the sort of high level thing was, oh, Redshift does this query in 30 seconds, and it takes Hive 10 minutes or something like that. But what they left out was that it had taken 17 hours for them to load the data into Redshift before they could query it. So it's very important to consider both, I think, time to query as well as time of query when you're talking about these problems. So another significant thing about foreign data wrappers is this is a public API. It means that we can use this as an extension point and continue to build on it without having to worry about forking the code. I mean, without foreign data wrappers, we could have defined this API on our own. But it would have been a very brittle thing to have to maintain. And I think more often than not, what you see, definitely you've seen this happen in the past with other companies that have distributed databases on top of Postgres. They end up basically forking pretty early on. One of our design goals from the very beginning has been to structure CitusDB in such a way that is decoupled enough from Postgres SQL that with each new version of Postgres SQL, we can, let's say, in a matter of weeks, rebase the changes that we've made onto the new version and then deliver to our customers both the value that we've added as well as the new features that the community has provided. And that's something that we're really excited about. Yes? So, I think that there are CCIs in the process of getting the data that we need to be able to maintain that kind of pattern. You know, that's probably a better question for my co-worker, Samid, to answer. Yes, good question. Okay. So let me just say that I love foreign data wrappers. I think that they are like the coolest thing ever, but if there's one thing I don't like about foreign data wrappers, it's the name, because I think that it seriously undersells the power of this API. And in particular, when you're trying to talk to a customer or an analyst and you use the word foreign data wrapper and you're like, yeah, this stuff is awesome. We're using foreign data wrappers. It allows us to do all these cool things. I mean, they hear two words like foreign and wrapper, which in this context don't necessarily sound that good, especially when you're talking about data locality and then you say foreign, I think, oh, you know, you're obviously like pulling data over the network and wrapper just sounds like some, you know, lightweight piece of stuff. Yes? SQL Med, yeah, would be a good option. But then the problem with SQL Med is you have to say like, oh, yeah, but that's like a 10 or 15 year old standard. And then that sounds boring, right? I mean, you want stuff to be new and you want it to have this cool buzzword attached to it. And unfortunately, anyway, we should have form a focus group and we'll work on this. It's a good idea. So what I've started saying is that really foreign data wrappers are a pluggable storage engine. It is an API for pluggable storage engines. And that's, to me, is what it looks a lot like and it's what it allows you to do. And one of the ways, in fact, in which foreign data wrappers are more powerful than HypeSurdees is that they allow you to do filter pushdown. And once you have the ability to do filter pushdown, it becomes very practical to build a foreign data wrapper on top of a column or format. So out of curiosity, by the way, Hans, how many people are familiar with like column or data formats? Okay, cool. So some people, but not enough that I shouldn't explain what I'm talking about. So traditional databases, when they lay data out on disk, they use a row major format, which means that if you're doing a table scan, you basically read row one, then row two, then row three, et cetera, et cetera. In a column major format, data is laid out like column one, column two, column three, or perhaps a chunk of column one, a chunk of column two, et cetera, and then the next chunk of column one, column two. So there are these sort of like hybrid formats like RC file or PAX where things are chunked like that. And the advantage of column or formats, especially for OLAP databases, where most queries are going to end up triggering full table scans, is that if you have a predicate that, let's be even more basic. Let's say that you're only looking at one of the columns in your query as opposed to doing a select star. You don't actually want to read those other columns off disk. I mean, we know that IO is pretty much like the biggest bottleneck. So if you can do a seek over just a large expanse of data on disk instead of doing a read as you would have to do with a column major format, that is like the single biggest performance benefit you can get right there. Another cool thing about column or formats is the impact they have on compression. So people have known for a while now that if you have an IO bound task, one way of improving the performance is actually to compress your data. And this is sort of a little bit counterintuitive because you think, oh, well, if I'm using compressed data, I have to devote CPU resources to doing the compression and decompression. But it turns out that the amount of time it takes to do that is much less than the amount of time that you save in terms of doing the IO to transfer the compressed data from disk into cache. And column or formats make this even better because it turns out that most of the compression algorithms that people use will actually generate higher compression ratios if the stuff that you're trying to compress is similar. So in other words, let's say that you have a string column, an int column, a string column, an int column. If you have to compress that data in a row major format, the compression ratio will not be as good as if you were, the other case, were using a column or format where you compress an entire int column followed by an entire string column. So this is one of the reasons why we're really big on foreign data wrappers. For us, the ability to be able to plug in multiple different column or formats seems like a huge win, especially in the context of Hadoop. Since right now, there are several different competing column or file format standards. And it's nice to actually have the option of supporting multiple formats as opposed to just one. So and then this is my final thought about foreign data wrappers. So someone in the audience mentioned SQLMED earlier, which was the original like ANSI SQL standard that sort of outlined this idea of adding this sort of like standard for creating a level of indirection between the tables that you see in one database and actually having them redirect to data that's stored somewhere else. And the use case that people were thinking about 10 or 15 years ago or whenever it was when they came up with this standard was really database federation. And the problem that they were trying to solve was that enterprise organizations don't have one database. They probably have 10 or 100 databases. And that creates a big management problem, especially when you really want to be able to have sort of a unified view of all of your data. So if you're able to create tables that behind the scenes actually redirect to data that's managed by another table, that's a huge sort of management win. But what SQLMED, at the time, I don't think anyone was really thinking about, was what this would look like in more of a distributed systems context. And so let's think about it this way. Let's suppose that you have a single PostgreSQL instance and you have a MongoDB foreign data wrapper. You are able, from your single PostgreSQL instance, to push a filter down to MongoDB, but anything that results from that sort of query then has to be pulled back across the network so that on your single node, let's say you can do the rest of the aggregation that you wanted to do or something like that. Now if you have a distributed version of PostgreSQL, you can actually co-locate the PostgreSQL worker processes on the nodes that are used by your distributed block store. Let's say MongoDB, HBase, or for example HDFS, and then you can both push the filter down into, let's say, MongoDB, but once you pull the results back up, instead of sending them over the network, you can apply part of the aggregation in a data local fashion. So the amount of IO that has to be done over the network is significantly reduced. So the point I'm trying to make, I guess, is that there's federation and then there's distributed federation and distributed federation is a pretty cool idea. Yes? Okay. So we don't support insert or updates in CitusDB, and that's part of being like an old app database. In terms of supporting transaction-oriented things in the underlying block stores, for example, MongoDB, I think, I don't know, what would you say? Actually, could we just save the questions for the end just so I can get through the, I have like two more slides and then we'll break for questions. Okay, one more point. So one of the other reasons why we decided to build CitusDB on top of PostgreSQL, I mean, there are a couple reasons, right? Like why reinvent the wheel when you can build on top of something that's already great? Another reason being that building a distributed database is more than just building a distributed query compiler and execution engine. It's also like all of these tools that sit around it. And I've definitely learned that the hard way from working on Hive, where you implement one thing and people are like, that's great, but what about these 10 other things that you haven't yet finished? But I think another thing that we've been really excited about is the fact that you have these two communities, which are both very vibrant, which are doing great work, and we really see CitusDB as a way of sort of combining those two communities into one, or at least providing people on either side with access to the great tools that are produced by the other. So for people who are already part of the Hadoop community, here's a chance to use an enterprise class database and an enterprise class feature set on top of Hadoop while maintaining the sort of characteristics of Hadoop in terms of scalability and cost effectiveness and for PostgreSQL users. Here's a chance to use the tools that you're already familiar with, but to use them on top of a platform that allows you to analyze petabytes of data. And also, in addition to that, we've sort of introduced this, I think, new level of flexibility with schema on read and features like that. So I just wanted to quickly sort of give some closing thoughts. So I think the first one is that it's a little bit unfortunate that we have this name Hadoop which lumps together MapReduce and HDFS, because I think as a result, people think that these two components are inextricably linked together and that they're of equal value. I mean, this is my opinion, but HDFS is much, much, much more valuable than MapReduce. MapReduce is not the only way to distribute computation across the data nodes in your cluster. I mean, CitusDB is one example of another way, right? Instead of using MapReduce, we have our own custom operator there and we get a more efficient system, a more expressive system as a result. And I think the other thing is that HDFS is really what is sort of driving industry adoption of Hadoop because of the way that it commoditizes storage and, you know, effectively sort of frees you from the clutches of the NAS and SAN mafia. And also, I mean, it's really changing the way that organizations manage data. I mean, the fact that it's so inexpensive compared to NAS and SAN technology means that this data that you were previously just writing to tape and effectively chucking, you're now keeping on site. And once you have it on site, of course, well, if we have it, why not query it? Why not try to extract some information from it? And that, you know, in turn is sort of driving this work to try to figure out how we can build, you know, databases that look familiar, that provide this familiar feature set on top of this new storage layer. And the storage layer is significantly different from the storage layer that already exists in the enterprise. And it's these differences that sort of require us to devise this new architecture. So it may seem like a little strong to say that, like, this is a new generation of analytic databases, but I do think, I genuinely think that they are significantly different from the databases that, you know, evolved over the previous decade. And I think that that, you know, description is warranted. So the other sort of closing thought just has to do with, you know, a summary of sort of the core benefits of SQL on Hadoop and more, you know, more specifically Postgres SQL on Hadoop. So, you know, in terms of sort of new features, things that weren't possible before, we have this nice ability to sort of decouple the logical schema from the underlying physical data. So you don't have to do loads. So you don't have to wait for your DBA to model your data before you can analyze it. So it has both, I think, an impact on the way that you interact with your coworkers and your ability to, like, quickly, you know, run queries against things. It also has a very significant impact on the way that you utilize your underlying storage. If you have to load data into a database, you're not just transforming it, you're also copying it. So you end up with multiple copies of the same data in multiple different storage systems across your data center. Doing schema on read means that you have one copy and you can even have the, you know, you can even project multiple different logical views onto that data as sort of an additional level of flexibility. Another core benefit is that since we're building on top of a horizontally scalable file system and leveraging, you know, the sort of key principle that makes it scalable data locality, we're able to maintain those same characteristics. The same goes for fault tolerance. And then I think the key thing that sort of differentiates this new generation of distributed databases from Hive is that by replacing MapReduce, we're able to achieve low latency, which means that you are now able to do interactive queries against small slices of data. And that, you know, once again really changes the way that you, you know, you can interact with your data. We're hiring. We really like PostgreSQL. We hack on PostgreSQL a lot. It seems like this is a pretty good audience, you know, for doing recruiting. So if any of you folks are interested in perhaps getting a job or changing jobs, you should definitely talk to us. That's a link for our website. We also have a bunch of interesting blog posts. And also another cool thing is you can actually download a copy of SiteSDB, install it on your machine and take it out for a test drive. We have EC2 images available, as well as I think what, CentOS and REL RPMs. Okay, questions? Yes? The license. So, actually, I mean, so parts of it are open source and parts of it are not open source. So all of the foreign data wrappers that we've written are open source. The code that we use for syncing metadata off of the Hadoop name node is open source. The parts that are closed source right now are basically the distributed query planner that we wrote. And I don't think that we, you know, speaking to my coworkers, I don't think we have any sort of like inherent objection to at some point in the future possibly open sourcing this. It's more of just to like, you know, we can't do that right now, right? That's how I'd explain that. Yes? There's a certain threshold of data amounting stuff where kind of usually when you're doing a closed-grass query versus a kind of additional network of a new power detector and close to all these various servers, you get kind of like an upper-parabyte. So I would say like if you're getting close to like, let's say, 10 terabytes of data, that's probably where this becomes, you know, a practical solution. Most cases like, I think we have two different types of customers. So one customer sort of like archetype are people who already have a Hadoop cluster setup. And they're using Hive, but they're like, man, I can't stand the latency or I really wish it, you know, provided better support for SQL or I already used Postgres SQL and now I'm using Hadoop and I really wish I could run, you know, one on top of the other. Then we have other customers who are, you know, doing things at, let's say like, terabyte scale, right? So they have, they're running Postgres SQL, but it's starting to sort of slow down a little bit in hiccup and they know that they need to go for a distributed solution, but they're not necessarily interested in setting up their own Hadoop cluster because that adds this additional management layer. I didn't mention it in the presentation, but CitusDB actually has two different deployment modes. So in one mode, you deploy it on top of Hadoop and HDFS basically manages the storage. In the other deployment model, every worker node basically runs Postgres SQL, but it runs on top of the native file system like ext3 or ext2, in which case it just uses Postgres SQL's storage engine for managing the table shows basically. We would not actually use Postgres foreign data wrappers. We would actually create basically like fragment tables on each one of the, on each one of the workers using basically like the size of these fragment tables is configured. And then when the user loads data into this sort of like distributed Postgres database, we sort of in parallel then stream data into these different fragment tables on the worker nodes. Yes. No, it allows. Push queries. Right, so it's basically up to the implementer of the foreign data wrapper to support things like that. So, you know, you have the ability right now to, let's say, extract filter conditions, which are something that are easy to push down, but you have to re-express that in terms of the API of the thing that you're talking to, in terms of the thing that you're building the wrapper on top of. So, sometimes it depends on whether you need to change the database. Change the runtime. Change the runtime. Okay, let's take this question offline. I'm not sure how to answer it right now. So, the question is if you have a complex query, can you gather all of the data to one local site? I'm not sure I understand why. What problem are you trying to solve by doing that? Okay, okay, sure. So for example, the join, right, a join that references multiple tables. So there are different strategies in distributed databases for doing joins. So one strategy that we support right now is a broadcast join, where if you're joining a small table for a large table, you can broadcast, basically, or copy the small table to each one of the worker nodes so that it has sort of a local copy. There are other strategies for doing big table joins in a distributed fashion that we're working on implementing right now. But the basic issue right when doing a big table join is that you do have to transfer data between the different worker nodes, because you can't expect that any given worker node will have all of the data locally stored that it needs in order to process like its fragment of the join. And the algorithms that have evolved for handling this distributed join case basically try to minimize the amount of data that you need to transfer. So there are different strategies. In our case, it's a little bit more complicated since we're not able to basically assume anything about the underlying data in each block. Since storage is completely delegated to HDFS, we can't do some of the optimizations that you see in other sort of more traditional MPP databases where they pre-sort or they hash partition the data across the nodes using a strategy like that. Makes sense? Okay. Question? Yeah. So you basically use our version of the copy command. So like I mentioned earlier, there are two different modes for operating CitusDBN. So if you're in mode one where you're actually using PostgreSQL's storage layer instead of using Hadoop, then you would use our distributed copy command. And you would basically say, here's the file, and I want you to copy it into the following table. And it would then, in a distributed fashion, basically stream that data into each one of the worker nodes. But in the typical sort of OLAP use case scenario, the data that you want to process is like fact data or it is log data. So it's not stuff that you're actually going to manipulate. You basically want to do joins or aggregations or analytic SQL queries on that information. And then in the case of deployment mode two where it's running on top of HDFS, you would use the Hadoop tools for basically loading data in. So Hadoop has something called distCP which is a distributed copy program, which is typically used there. Any other questions? Yes. I'm going to actually let Samir answer that because he's been working on this recently.. Yes. When Calculator minimum delivery of an incoming fl거h micromoji is nice they have obviously This version we can for the Hadoop version we can't. Any other questions in the back? S och simil It seems to me that the Yes, Krishna? So we're using that at the master node level in order to replicate the table metadata that the master node maintains. So this is basically information that maps a global table name to all of the foreign tables that exist on each one of the worker nodes. So it's basically the table metadata, both at sort of the global master level as well as the map, thanks from the global level to the worker level. Which is what we want to do, because there's a single master or a single active master, but you could have streaming replication multiple hot standbys. And we're not interested in replicating, in a sense, real data, because no real data exists on the master. It's all metadata about basically where the shard records are stored on each one of the worker nodes. Any other questions? Okay. Thank you.
 My name is Carl Steinbach.  I'm an engineer at Citus Data.  I'm also a member of the PMC and a committer  on the Apache Hive project, which is kind of like a database  project related to Hadoop.  And in the past, I've worked at a bunch of enterprise software  companies, most recently, Cloud Era, where  I was team lead for the Hive team.  I've also spent time at Informatica, NetApp, where  I worked on storage security products, and Oracle,  where I started my professional career working  on the database server.  My other claim to fame is that I wrote that blog post about  Tom Lane's email habits, which I think a lot of you  probably read.  Thank you.  Thank you very much.  So hopefully, this presentation lives up to that  blog post.  Anyway, I've given talks on this topic a couple times in  the past.  And from those experiences, I've learned that it's a good  idea to start upfront by making sure that we're all on the  same page about what a distributed  analytic database is.  I think probably 95% of people, when you say relational  database, they automatically think OLTP, the standard  sort of create, update, delete, use case scenario with  transactions.  And an analytic database is definitely not OLTP.  It's typically referred to as an OLAP database.  So the kind of things that you're going to do with an  OLAP database include processing, let's say,  clickstream events, or processing log files, or  processing some kind of event or fact type data.  And typically, people are going to do aggregations on  those data or big joins.  But in terms of the way that the actual database behaves,  one thing that sort of defines OLAP use cases is large  sequential scans, where typically you're looking at  the entire table as part of a query.  Another thing is they're not really transaction oriented.  So a lot of the stuff that you see in OLTP databases don't  really apply.  And then I think finally, and this is probably most  important to the rest of this presentation, in terms of the  performance characteristics of OLAP databases, they are  typically IO bound.  So if you're looking for sort of the performance bottleneck in  the system, it's not going to be the CPU.  It's typically going to be the rate at which the database  process can actually pull information off of disk.  So a lot of work over the past 10 or 20 years has gone into  like figuring out how to reduce the impact of that  bottleneck or eliminate it altogether.  So one thing that also relates to what we're going to  discuss is the enterprise storage model.  And this is something that's evolved over, let's say, the  past 20 or 30 years.  And the goal of this is to basically solve the problem  of availability and accessibility for data in the  context of, let's say, a large enterprise, a company with  10, 100, 1,000 employees, and terabytes, and possibly even  petabytes of data.  So obviously, that data is not going to sit on one computer.  It's going to sit on a bunch of computers.  And you want somehow to make it accessible to a large number  of people in your organization.  So you need to be able to access it over a  network of some sort.  And the sort of solution that's evolved to this over time  and really that a whole industry has been built around is  basically structuring your data center such that you have a  pool of storage.  And this could be in the form of a NAS or a SAN.  And then it's connected via some kind of switching network  fabric to a pool of compute resources.  So there are a couple problems with this.  One problem that customers feel immediately is cost.  This is a really expensive solution.  This is a lot more expensive than going out and buying a  hard drive at Fry's Electronics or Best Buy or  something like that.  I mean, you're talking about spending upwards of $1,000  per terabyte.  Probably even more conventionally.  And then add to that the fact that these systems have  sort of a built-in scalability problem.  And the scalability problem is that really big pipe that  sits between your island of compute resources and your  island of storage resources.  So as you attempt to scale out your storage pool, you're  going to have to make that pipe bigger and bigger and  bigger.  And in practice, I mean, there are sort of limits to how  big you can make that pipe.  There are also limits to how consequently how much you can  scale the underlying storage tier.  And what happens as a consequence is that people  tend to sort of partition these resources, which means  they'll end up with multiple islands of storage pools and  multiple islands of compute workers.  And then that adds, on top of it, an additional  manageability problem.  And there's sort of a whole industry that's grown around  just solving that, helping you to shuttle data from one  storage pool to another to meet SLAs or jobs that you have  running in the background.  I also want to say that this diagram is sort of a gross  oversimplification.  I'm trying to get the main points home.  But I wanted to point out that besides just having a  storage pool, there are actually, in a sense, multiple  different tiers of storage pool with different price  performance characteristics.  So oftentimes, in the compute pool, you might have an OLAP  database, and that database will have its own sort of  private storage pool, which will often be, in terms of  cost per terabyte, much more expensive than sort of your  larger pool of online storage.  And consequently, you fall into this usage pattern where  you're always sort of moving data between these two pools.  Because you can't fit all of the stuff that you want to look  at in the sort of fast, small pool of storage.  And as a result, there are pretty big management headaches.  So over 10 years ago at this point, the folks at Google  were building this gigantic search engine.  And in order to build the inverted index that they  needed to power that search engine, they were collecting  these gigantic data sets by doing web crawls.  And I think they realized pretty early on that if they  got hooked on sort of the NAS or SAN solution, that that  would not be just a bottleneck for their data center, but  it would really be a bottleneck for their business.  So they knew that they had to find a different solution to  that problem.  And they knew that whatever solution they ended up with,  it had to both be inexpensive.  And it had to be very scalable.  So the sort of design priorities that they set out  were, first off, we're going to base this on commodity  hardware.  We want the hardware that the system is built on top of to  be as inexpensive as possible.  And sort of a consequence of that is when you go with  inexpensive commodity hardware, you expect the stuff to  fail all the time.  And if you're deploying 1,000 of these boxes, you expect  probably one box to fail every day, if not multiple boxes.  So you have to have a system that can be resilient in the  face of these constant failures and outages.  So another really interesting thing about this is that the  solution that they came up with, of course, is referred to  as the Google File System.  But in many ways, it's not really a  conventional file system.  It's more of, let's say, a distributed block store.  And there's an interview online that you can find with  Sean Quinlan, who was one of the original architects of this  system.  He sort of talks about the genesis of how they came up  with this sort of set of priorities and the things that  they decided were not so important.  And I think one of the really critical things is that they  weren't worried about supporting or building a  POSIX file system.  They weren't worried about supporting random reads or  random writes.  So there's a lot of stuff, a lot of file system dogma that  they just dispensed with very quickly.  And in the interview with Quinlan, he makes the point  that since the people who were building GFS were actually  the same people who were building the system on top  of GFS that were consuming it, that they were able to be  very fluid in terms of defining what the requirements  were for this system.  And I think that's kind of, it's an interesting thing to  think about in terms of how we build systems in the future.  That sometimes just these sort of little political  disagreements between groups at different layers in the  system can really stymie advances like this.  So the system that they ended up with was referred to as the  Google File System.  And this is sort of a high level overview of how it works.  So when a file is written to GFS, it's basically split up  into a number of fixed size chunks.  So in the original GFS implementation, these chunks  were 64 megabytes in size, which is a lot larger than the  blocks that you find in a traditional file system.  But it makes sense in the context of storing files  that on average are going to be gigabytes, if not  potentially even terabytes in size.  And the advantage, of course, of having big blocks, or at  least one really significant advantage, is that it reduces  the metadata load on whatever system you're using to store  the file system metadata.  There's less of it to store since the number of objects  that it needs to track and keep track of are limited in  number.  So another sort of significant design element here is that  once that file is split up into chunks, those chunks are  then parceled out more or less randomly to a set of  chunk servers.  And three copies, or whatever the replication factor for  the file system are, by default, three copies would  then be distributed across the chunk servers in the system.  So there are, in one sense, this is very different from the  way that traditional distributed file systems  achieve reliability by using something like RAID or a  parity-based striping mechanism, which is more  efficient in terms of the amount of disk space that's  used.  I mean, you're able to, let's say, if you use RAID 5,  you're basically trading off a fifth of your storage space  for that additional reliability.  So this is a lot less efficient in terms of disk space, but  because they're actually creating multiple copies of  every chunk, they do get some other benefits in return.  So one of them is that the load can be distributed very  evenly across these chunk servers when the same file is  being read by multiple clients.  So basically, a coordinator has the option of saying, if  these three clients are reading the same file, they  don't necessarily have to all talk to the same chunk server.  Another advantage of this is that in the system that they  devised, they expected not just disk failures, but they  also expected to have chunk server failures.  They expected the whole node to just disappear and go down.  And at that point, if you're basically building chunk  servers on top of RAID, the RAID wouldn't matter, right?  Because you lose the chunk server as an entire unit.  So this way, when a chunk server goes down, you have two  other chunk servers available where you can read the same  block from.  And finally, the other nice characteristic of this is that  when a chunk server goes down, since all of the blocks  that are on that one chunk server have replicas  distributed in a more or less uniform fashion over the  remaining chunk servers, the load that's required to read  those chunks back and copy them to a new chunk server in  order to maintain the replication factor is then  distributed evenly across the other servers in the pool.  So earlier, I mentioned that the problem that Google was  contending with was twofold.  They were worried about the cost of the traditional NAS and  SAN solution, and they were also worried about scalability.  So I think if you look at just the stuff that's described in  the original GFS paper, and you hold it against these goals,  based on what they articulated in that paper, it's pretty  clear that they solved the cost problem.  I mean, they articulated a way of building a very scalable  file system using commodity components.  But I don't think that if you look at just the GFS paper,  that they really solve the scalability problem.  Because fundamentally, it's still a client server type of  file system, where the clients are all going to pull data  across the network.  And that network is then going to become the  IO bottleneck.  And it wasn't until a year later when they published the  MapReduce paper that the missing puzzle piece appeared.  And it was in the MapReduce paper where they said, no,  actually, here's how GFS works.  This is not a conventional file system where you have clients  are pulling data over to the client in order to do work on  the data there.  Instead, we are going to take the work and push it over to  the data.  So MapReduce is basically a distributed computing  framework that allows you to send work over to the nodes  where the data actually exists.  And by maintaining data locality, you almost completely  eliminate this network bottleneck problem.  MapReduce does require some network traffic between nodes.  It has a shuffle and sort phase and stuff like that.  But at least at a first order, you've basically eliminated  this really significant bottleneck.  And that, I think, is the key benefit of Hadoop and of HDFS.  And I think it's really interesting that if you just  look at the GFS paper, that this missing piece is not  mentioned anywhere there.  But if you go back and you read that paper, you can tell  that that's what they're thinking.  It's kind of sort of an interesting historical thing.  So these papers were published in 2003 and 2004.  And at around that same time, Doug Cutting and Mike  Cafferella were working on an open source project called  Nutch, which was an outgrowth of Apache Lucene.  And their goal was to build an open source search engine.  So they saw these papers and they realized immediately,  hey, this is clearly the way that we have to build Nutch.  We need to build a distributed file system equivalent to GFS.  And we need to devise our own map reduce  implementation.  And that's what became Hadoop.  So Hadoop is these two things.  HDFS, which is Hadoop's version of GFS, and then their own  map reduce implementation.  So that project came into existence in 2005.  It was initially used quite heavily at Yahoo and then  Facebook.  And in the last, I would say, three to four years, we've  seen sort of more general adoption within enterprise  data centers.  So my previous employer, Cloudera, has sort of built a  business around helping companies that are, let's  say, already using Oracle or already using SQL Server or  something like that to figure out ways of sort of  transitioning to this lower cost and more scalable  infrastructure.  But the fact that people are actually using this in  industry has given us an opportunity to get a sense for  what are the practical benefits of Hadoop and also what are  the drawbacks.  So in terms of benefits, I think at the very top of the  list is the way that HDFS has commoditized enterprise  storage.  And this has had really significant effects on in  terms of what enterprise organizations are doing with  their data and also the amount of data that they're  actually retaining on site.  So in the past and actually still today, you find that  most large organizations have some data that they maintain  in an online fashion in their data center.  But a lot of the data that they collect is getting  written to tape and put on a truck.  And the truck drives to a cave somewhere.  And it's deposited there.  And no one ever looks at it again.  And some of you guys might think I'm joking about the cave  thing, but you should look up Iron Mountain, which is one  of these companies.  They literally have a cave.  So that's one major thing.  Like companies that were formerly putting a lot of  data on tape never to be seen again are now maintaining it  in HDFS instead.  And they're able to derive a lot of benefit from that data.  Another big advantage is the way that HDFS allows  companies to scale out in a genuinely fashion instead of  having to invest in more expensive hardware to scale up.  It also gives them the ability to run a system in a  fault-tolerant manner, I think in a manner that wasn't  really available before with other existing NAS and  SAN solutions.  And it's also introduced a new level of flexibility in  terms of the data that these organizations can operate on.  So one of the things about Hadoop is that it is, or HDFS  in particular, it's a write-once-read-many-times file  system.  It's like an online archive file system.  So the ability to operate on data in place in the format in  which it lands on disk is very attractive.  And Hadoop offers flexible tools that allow you basically to  extract information from a variety of different formats.  So at the same time, I think people are aware that Hadoop  has a bunch of drawbacks.  And chief among these is MapReduce itself.  MapReduce is very powerful in the sense that it's extremely  flexible, but a lot of that flexibility comes from the  fact that it's so generic or so general and also so low level.  So to do even sort of simple operations, you're going to  end up writing pages of Java code.  And once you've written this code, it's more or less  impossible to maintain.  I mean, you're going to have trouble remembering a month  later what the code you wrote does and your ability to share  it with a coworker is close to zero.  So part of this relates to the fact that MapReduce doesn't  have any schemas.  And schemas are really valuable.  I think many times when people think about schemas in the  context of a relational database, they maybe think that  in some ways schemas are the enemy.  Because oftentimes you'll find that maybe you have some  data that you really want to analyze in the database.  But before you can even get it into the database, you have  to talk to your DBA.  And your DBA is going to spend a couple of days trying to  figure out the right schema for this data.  So there's a lot of organizational latency built  into this.  And the reason for that is that if you get the schema wrong,  that's an expensive mistake to make.  So schemas in the traditional context are a high stakes game.  But they also provide a lot of value.  Because they allow you to view data in a logical fashion as  opposed to having to worry about the underlying physical  structure of it.  And Hadoop, without schemas, forces you to constantly worry  about the underlying physical structure.  And then the other missing features I think can quickly  be summarized by saying that people really want Hadoop to  look more like a database.  They want Hadoop, they like the scalability, they like the  flexibility, but at the same time, databases look the way  they do for a reason.  They've evolved over the past 20 or 30 years in response to  the needs of users.  And they provide a bunch of features that make a lot of  sense when what you're trying to do is analyze data.  So these include things that help you write code or help you  write better code, like optimizers, indexes, and views.  And there's also this large ecosystem of tools that run on  top of databases that know how to speak things like ODBC and  JDBC, and which allow you to do really interesting things  with your data, like business intelligence tools, ETL  tools for transforming and manipulating your data.  And then things like SQL IDEs or other environments for  doing data analysis.  So the way that the Hadoop community responded to these  shortcomings of the sort of core platform was by building a  variety of domain-specific languages on top of Hadoop.  So the idea was you no longer have to code your job in  MapReduce, instead you can use this other language which we  compile down to MapReduce.  So there were a variety of languages that evolved like  this.  Google, of course, has something called SAASOL, which  they published a paper on.  They also have a SQL DSL referred to as Tensing.  Within the Hadoop world, we have things like PIG.  We also have Cascading, which is kind of like a Java data  flow API.  The DSL that I think has gained the most interest and  adoption within the enterprise software community is  Hive.  So Hive made the very pragmatic decision instead of  inventing our own language, we're going to stick with  SQL.  Because SQL has some rough edges.  It has some bugs.  But at least we've learned what those bugs are over the  last 30 years, and they're easy to work around.  And also, it turns out the entire world basically speaks  SQL, including all of those ecosystem tools that I  mentioned earlier.  So SQL is like the Esperanto of data processing, except  that people actually speak it, unlike Esperanto.  So anyway, that's the situation we're in.  So one thing about Hive that I really want to try to  convey is that most people, when you mention it, they  think, OK, Hive is like a SQL to map reduce  compilation and execution engine, which is very true.  That's what it is.  But it's also these two other things that I don't think it  gets enough credit for being.  So one of those things basically is the data model  that it provides.  And by data model, I mean the way that it basically allows  you to map logical structure in the form of schemas to  underlying physical data that can be in basically any  format you want.  So in order to provide that functionality, Hive has two  components.  One is the Hive Metastore, which basically allows you to  think of it as Hive's table catalog.  So the Metastore maintains mappings of the form table  basically to underlying file in HDFS, or to HDFS location  where the data is stored.  And it also maintains mappings of table to SIRT name, where  SIRTs are basically these plug-in libraries that allow you  to handle different formats.  So with these two bits of information, when Hive  executes a query plan, which is in the form of basically a  graph of map reduce jobs, each one of these map reduce  jobs gets sent to a data node, which you could think of as  HDFS's version of chunk servers in GFS.  And the map reduce job, which is expressed in terms of Hive  operators, is executed there.  But the data, as it's read off of HDFS, is filtered through  one of these SIRTs.  So an example would be, let's say, a CSV SIRT, or a JSON  SIRT, or perhaps an RC file SIRT, which is Hive's  columnar format.  But in this way, it allows you to do something really cool,  which I think people have started to refer to as schema  on read as opposed to schema on write.  So this is the really cool benefit of Hive, and the  thing that I expect will be Hive's lasting impact on the  database world, not SQL to map reduce, but this very  flexible way of mapping logical structure onto  underlying physical data.  Yep?  So basically, a SIRT, by the way, SIRT stands for  serialized, deserialized.  So SIRT, S-E-R-D-E-F.  So when you are deserializing, you are basically reading  from a file, and what you are pulling out of it are rows  split into columns with types.  And in the reverse direction, you feed in basically these  structures that are typed columns, and then it writes  it out to the underlying file.  So you were saying, could you have a SIRT that goes from  CSV to, let's say, JSON, or something like that?  You'd actually use two SIRTs for that process.  You would use a JSON SIRT and a CSV SIRT, and what you would  do would be to construct two tables, one of which is based  on top of JSON, the other on top of CSV.  And then one interesting thing about Hive is that it's  added a variety of sort of extensions to base SQL,  including semantics that allow you to easily do things like  stream data from one table into another table.  So I hope that answers your question.  So I think Hive solved a bunch of problems, but in many  ways it's also been sort of a victim of its own success.  Because as soon as you start making something look like a  database, people are like, hey, this is great.  I like that it looks like a database, but I wish you would  cover that remaining 50% and make it look completely like a  database.  Where are this missing SQL feature that I'm used to  using?  Where is the authentication and authorization system?  Here's a data type that I expected you to support.  It's part of the SQL standard, but I don't see it here.  And then another problem with Hive from the standpoint of  many users is caused by the fact that it's basically  delegating execution to MapReduce.  So MapReduce is a very powerful tool, but it makes  some compromises which are probably not the best  compromises to make in the context of a  database engine like Hive.  And the big compromise that it makes is that  it's batch oriented.  So it guarantees basically really high throughput.  But it has this built in latency as a result of the  fact that when you start a job, you have to talk to the  job tracker.  And the job tracker has to go and parcel out jobs to each  one of the data nodes so that as a result, even if you are  running a query against just a small sliver of data, let's  say a megabyte or 100 megabytes or something like  that.  The minimum turnaround time on that query is going to be  15 to 30 seconds.  And if you ran the same query, let's say, on Postgres SQL  on a single node, you'd expect it to complete  in under a second.  So with Hive, this whole way of interacting with a database  where you do very small exploratory queries to try  to gain an understanding of the layout of your data and  what's actually there.  It's not really feasible or practical with Hive.  So in the past year to maybe two years, I think it's fair  to say a new generation of distributed analytic databases  has emerged, which are largely being designed to solve  this set of problems.  And in particular, the problem of the latency overhead and  inefficiencies that are introduced by using MapReduce.  So examples of these systems would include things like  Impala from Cloudera, the Apache Drill Project, Google's  Dremel as an example of this, also within Google, and  CitusDB is another example of that.  So I'm going to talk a little bit about CitusDB.  But before I do, I wanted to show the common architectural  difference that sets these new databases apart from, let's  say, the previous generation of MPP databases that emerged  over the previous decade.  So this looks a lot like what we've seen recently where  people have been adapting MPP databases to run on top of  Hadoop.  And what they're effectively doing is creating a  connector between the worker nodes in the MPP cluster and  the data nodes in HDFS.  And the problem with this approach is that you have the  IO bottleneck between these two layers.  So you're not achieving data locality for your jobs, and a  system like this is just not going to scale.  But the solution is, at least if we want to illustrate it  graphically, it's pretty straightforward.  You get rid of the dedicated MPP data nodes, and you take  the processes, the worker processes that were running  on those nodes, and you co-locate them on  each one of the data nodes.  So in effect, what you're doing then is pushing the work  down to the data as opposed to pulling the data over to the  work.  And this allows systems like this to scale out in the same  way that you're able to scale out with  something like MapReduce.  So I wanted to go into some more detail about how this  actually plays out in a system like CitusDB.  So CitusDB is built on top of PostgreSQL.  And if we look at the components in the system, on  the left-hand side, we have PostgreSQL clients, ODBC and  JDBC driver.  And I think this is one of the really significant points I  wanted to make, that as far as a client is concerned, as far  as a user is concerned, CitusDB looks almost identical to  PostgreSQL.  So you basically continue to use PSQL or the different  tools that you're familiar with.  The support for SQL is virtually the same.  All of the same system tables are there and stuff like that.  But in terms of changes we've made, on the master node,  once again, it's more or less stocked PostgreSQL with the  addition of our distributed query compiler and  execution engine.  And then on each one of the HDFS data nodes, we co-locate  what is in effect a stock copy of PostgreSQL, along with a  foreign data wrapper that we've written that allows us to  basically read from HDFS.  So I wanted to go into a little bit more detail about how  this actually works by giving an example of how  query is executed.  So to start off, when you create a table on the  CitusDB master, you would basically specify a directory  or a file in HDFS that that logical table name needs to  map to.  And then as part of a prior offline step, the master node  syncs file system metadata off of the HDFS name node.  So this basically allows the master node to map an HDFS  file name to a set of blocks and data nodes in the  underlying storage pool.  And then, once the table is created and once the metadata  has been synced, the master node talks to each one of the  worker nodes, and it creates a foreign table per block on  each one of the data nodes using the HDFS foreign data wrapper.  So then subsequently, let's say the user submits a query, a  simple aggregation query to find the average salary of  managers in the organization.  So as a first step, we take that global query and we  rewrite it as a set of queries or fragment queries that  can be executed on the worker nodes.  And those queries then, or there's effectively one query  per foreign table where each foreign table maps to a block  on each one of those data nodes.  The worker nodes execute those fragment queries, generate  partial result sets, which are then returned to the master  node, which merges those results together and returns a  final answer to the client.  So there are some interesting details that I wanted to  cover at a slightly lower level.  So one of them has to do with block awareness.  So I had that previous slide where I showed that, in a  sense, what we're doing is taking a traditional MPP  database architecture.  And the most significant change we're making is that we're  co-locating the worker processes on the data nodes.  But I don't want to leave you with the impression that  that's all you have to do in order to achieve data locality.  Because the problem is that if that alone is what you do,  those MPP worker processes running on the data nodes are  still going to have to basically fetch data from all of  their neighbors in order to satisfy any query.  So it's very important that the master node, when it is  scheduling a query, when it is scheduling those fragment  queries, that it sends them to the correct data nodes so  that those data nodes can do those reads locally.  And that's what we refer to as block awareness.  So another thing that's worth mentioning is fault  tolerance.  So our solution to fault tolerance, I think, can  quickly be summarized by saying that we're basically  leveraging the best features of the two platforms that we're  building on top of, Hadoop and PostgreSQL.  So what happens if the master node fails?  Well, we take advantage of PostgreSQL streaming  replication to make sure that the table metadata is copied  over to a hot standby.  And that solves the problem at the master node level.  And at the worker nodes, since we are actually delegating  the storage problem as well as the fault tolerance  problem to HDFS, if one of those nodes fails, we already  know that the actual data has been replicated to a bunch  of neighbors and can quickly fail over to them.  And the missing piece, the piece that we actually had to  implement on our own, is the ability to, if a process or  if a query is in the process of executing, detect that a node  which is responsible for some of these fragment queries has  gone down and to reassign those queries dynamically to other  replicas in the system.  So one of the things that I think, yep?  It's the last of the replicas.  Right.  Then you're screwed, basically.  Yeah.  You would get an error back.  That's true.  And you would get an error back quickly because it would  determine that it can't communicate with any of the  three replicas in that situation.  So there is a limit to, yeah, no, fair question.  So back when I was first talking to the guys at Citus  Data about joining the company, they were talking about  what they had done with CitusDB and how they had built it  on top of PostgreSQL.  I wasn't very familiar with PostgreSQL at the time.  I was mainly familiar with things like MySQL and in  particular with things like Hive.  And as they were telling me about some of the different  features of PostgreSQL that they were leveraging, they  mentioned foreign data wrappers.  And I got really excited about this because foreign data  wrappers, I quickly realized, are really a much more  powerful version of Hive-SERDs.  And using foreign data wrappers, we are able to  achieve the same novel benefits that separate Hive from  sort of traditional databases.  The ability to do schema on read as opposed to schema on  write, which introduces this whole new level of  flexibility.  And very significantly, schema on read is more than about  just flexibility.  It's also, in a sense, about performance.  Because if you can read data in place, regardless of the  format, without first having to go through this process of  loading it, in other words, translating it or converting  it into the native format of the database before you can  read it, that's a significant performance in terms of  overall latency.  And just as a quick side note, a couple weeks ago or maybe  a month ago, someone posted a performance comparison between  Hive and Amazon Redshift.  And this was carried on Hacker News.  And the sort of high level thing was, oh, Redshift does  this query in 30 seconds, and it takes Hive 10 minutes or  something like that.  But what they left out was that it had taken 17 hours for  them to load the data into Redshift before they could  query it.  So it's very important to consider both, I think, time  to query as well as time of query when you're talking  about these problems.  So another significant thing about foreign data wrappers is  this is a public API.  It means that we can use this as an extension point and  continue to build on it without having to worry about  forking the code.  I mean, without foreign data wrappers, we could have  defined this API on our own.  But it would have been a very brittle thing to have to  maintain.  And I think more often than not, what you see, definitely  you've seen this happen in the past with other companies  that have distributed databases on top of Postgres.  They end up basically forking pretty early on.  One of our design goals from the very beginning has been to  structure CitusDB in such a way that is decoupled enough  from Postgres SQL that with each new version of Postgres  SQL, we can, let's say, in a matter of weeks, rebase the  changes that we've made onto the new version and then  deliver to our customers both the value that we've added as  well as the new features that the community has provided.  And that's something that we're really excited about.  Yes?  So, I think that there are CCIs in the process of  getting the data that we need to be able to  maintain that kind of pattern.  You know, that's probably a better question for my  co-worker, Samid, to answer.  Yes, good question.  Okay.  So let me just say that I love foreign data wrappers.  I think that they are like the coolest thing ever, but if there's one thing I don't like  about foreign data wrappers, it's the name, because I think that it seriously undersells  the power of this API.  And in particular, when you're trying to talk to a customer or an analyst and you use the  word foreign data wrapper and you're like, yeah, this stuff is awesome.  We're using foreign data wrappers.  It allows us to do all these cool things.  I mean, they hear two words like foreign and wrapper, which in this context don't necessarily  sound that good, especially when you're talking about data locality and then you say foreign,  I think, oh, you know, you're obviously like pulling data over the network and wrapper  just sounds like some, you know, lightweight piece of stuff.  Yes?  SQL Med, yeah, would be a good option.  But then the problem with SQL Med is you have to say like, oh, yeah, but that's like a 10  or 15 year old standard.  And then that sounds boring, right?  I mean, you want stuff to be new and you want it to have this cool buzzword attached to  it.  And unfortunately, anyway, we should have form a focus group and we'll work on this.  It's a good idea.  So what I've started saying is that really foreign data wrappers are a pluggable storage  engine.  It is an API for pluggable storage engines.  And that's, to me, is what it looks a lot like and it's what it allows you to do.  And one of the ways, in fact, in which foreign data wrappers are more powerful than HypeSurdees  is that they allow you to do filter pushdown.  And once you have the ability to do filter pushdown, it becomes very practical to build  a foreign data wrapper on top of a column or format.  So out of curiosity, by the way, Hans, how many people are familiar with like column  or data formats?  Okay, cool.  So some people, but not enough that I shouldn't explain what I'm talking about.  So traditional databases, when they lay data out on disk, they use a row major format,  which means that if you're doing a table scan, you basically read row one, then row  two, then row three, et cetera, et cetera.  In a column major format, data is laid out like column one, column two, column three,  or perhaps a chunk of column one, a chunk of column two, et cetera, and then the next  chunk of column one, column two.  So there are these sort of like hybrid formats like RC file or PAX where things are chunked  like that.  And the advantage of column or formats, especially for OLAP databases, where most queries are  going to end up triggering full table scans, is that if you have a predicate that, let's  be even more basic.  Let's say that you're only looking at one of the columns in your query as opposed to  doing a select star.  You don't actually want to read those other columns off disk.  I mean, we know that IO is pretty much like the biggest bottleneck.  So if you can do a seek over just a large expanse of data on disk instead of doing a  read as you would have to do with a column major format, that is like the single biggest  performance benefit you can get right there.  Another cool thing about column or formats is the impact they have on compression.  So people have known for a while now that if you have an IO bound task, one way of improving  the performance is actually to compress your data.  And this is sort of a little bit counterintuitive because you think, oh, well, if I'm using  compressed data, I have to devote CPU resources to doing the compression and decompression.  But it turns out that the amount of time it takes to do that is much less than the amount  of time that you save in terms of doing the IO to transfer the compressed data from disk  into cache.  And column or formats make this even better because it turns out that most of the compression  algorithms that people use will actually generate higher compression ratios if the stuff that  you're trying to compress is similar.  So in other words, let's say that you have a string column, an int column, a string column,  an int column.  If you have to compress that data in a row major format, the compression ratio will not  be as good as if you were, the other case, were using a column or format where you compress  an entire int column followed by an entire string column.  So this is one of the reasons why we're really big on foreign data wrappers.  For us, the ability to be able to plug in multiple different column or formats seems  like a huge win, especially in the context of Hadoop.  Since right now, there are several different competing column or file format standards.  And it's nice to actually have the option of supporting multiple formats as opposed  to just one.  So and then this is my final thought about foreign data wrappers.  So someone in the audience mentioned SQLMED earlier, which was the original like ANSI  SQL standard that sort of outlined this idea of adding this sort of like standard for creating  a level of indirection between the tables that you see in one database and actually  having them redirect to data that's stored somewhere else.  And the use case that people were thinking about 10 or 15 years ago or whenever it was  when they came up with this standard was really database federation.  And the problem that they were trying to solve was that enterprise organizations don't have  one database.  They probably have 10 or 100 databases.  And that creates a big management problem, especially when you really want to be able  to have sort of a unified view of all of your data.  So if you're able to create tables that behind the scenes actually redirect to data that's  managed by another table, that's a huge sort of management win.  But what SQLMED, at the time, I don't think anyone was really thinking about, was what  this would look like in more of a distributed systems context.  And so let's think about it this way.  Let's suppose that you have a single PostgreSQL instance and you have a MongoDB foreign data  wrapper.  You are able, from your single PostgreSQL instance, to push a filter down to MongoDB,  but anything that results from that sort of query then has to be pulled back across  the network so that on your single node, let's say you can do the rest of the aggregation  that you wanted to do or something like that.  Now if you have a distributed version of PostgreSQL, you can actually co-locate the PostgreSQL  worker processes on the nodes that are used by your distributed block store.  Let's say MongoDB, HBase, or for example HDFS, and then you can both push the filter  down into, let's say, MongoDB, but once you pull the results back up, instead of sending  them over the network, you can apply part of the aggregation in a data local fashion.  So the amount of IO that has to be done over the network is significantly reduced.  So the point I'm trying to make, I guess, is that there's federation and then there's  distributed federation and distributed federation is a pretty cool idea.  Yes?  Okay.  So we don't support insert or updates in CitusDB, and that's part of being like an old app  database.  In terms of supporting transaction-oriented things in the underlying block stores, for  example, MongoDB, I think, I don't know, what would you say?  Actually, could we just save the questions for the end just so I can get through the,  I have like two more slides and then we'll break for questions.  Okay, one more point.  So one of the other reasons why we decided to build CitusDB on top of PostgreSQL, I  mean, there are a couple reasons, right?  Like why reinvent the wheel when you can build on top of something that's already great?  Another reason being that building a distributed database is more than just building a distributed  query compiler and execution engine.  It's also like all of these tools that sit around it.  And I've definitely learned that the hard way from working on Hive, where you implement  one thing and people are like, that's great, but what about these 10 other things that  you haven't yet finished?  But I think another thing that we've been really excited about is the fact that you  have these two communities, which are both very vibrant, which are doing great work,  and we really see CitusDB as a way of sort of combining those two communities into one,  or at least providing people on either side with access to the great tools that are produced  by the other.  So for people who are already part of the Hadoop community, here's a chance to use an  enterprise class database and an enterprise class feature set on top of Hadoop while maintaining  the sort of characteristics of Hadoop in terms of scalability and cost effectiveness and  for PostgreSQL users.  Here's a chance to use the tools that you're already familiar with, but to use them on  top of a platform that allows you to analyze petabytes of data.  And also, in addition to that, we've sort of introduced this, I think, new level of  flexibility with schema on read and features like that.  So I just wanted to quickly sort of give some closing thoughts.  So I think the first one is that it's a little bit unfortunate that we have this name Hadoop  which lumps together MapReduce and HDFS, because I think as a result, people think that these  two components are inextricably linked together and that they're of equal value.  I mean, this is my opinion, but HDFS is much, much, much more valuable than MapReduce.  MapReduce is not the only way to distribute computation across the data nodes in your  cluster.  I mean, CitusDB is one example of another way, right?  Instead of using MapReduce, we have our own custom operator there and we get a more efficient  system, a more expressive system as a result.  And I think the other thing is that HDFS is really what is sort of driving industry adoption  of Hadoop because of the way that it commoditizes storage and, you know, effectively sort of  frees you from the clutches of the NAS and SAN mafia.  And also, I mean, it's really changing the way that organizations manage data.  I mean, the fact that it's so inexpensive compared to NAS and SAN technology means that  this data that you were previously just writing to tape and effectively chucking, you're now  keeping on site.  And once you have it on site, of course, well, if we have it, why not query it?  Why not try to extract some information from it?  And that, you know, in turn is sort of driving this work to try to figure out how we can  build, you know, databases that look familiar, that provide this familiar feature set on  top of this new storage layer.  And the storage layer is significantly different from the storage layer that already exists  in the enterprise.  And it's these differences that sort of require us to devise this new architecture.  So it may seem like a little strong to say that, like, this is a new generation of analytic  databases, but I do think, I genuinely think that they are significantly different from  the databases that, you know, evolved over the previous decade.  And I think that that, you know, description is warranted.  So the other sort of closing thought just has to do with, you know, a summary of sort  of the core benefits of SQL on Hadoop and more, you know, more specifically Postgres  SQL on Hadoop.  So, you know, in terms of sort of new features, things that weren't possible before, we have  this nice ability to sort of decouple the logical schema from the underlying physical  data.  So you don't have to do loads.  So you don't have to wait for your DBA to model your data before you can analyze it.  So it has both, I think, an impact on the way that you interact with your coworkers  and your ability to, like, quickly, you know, run queries against things.  It also has a very significant impact on the way that you utilize your underlying storage.  If you have to load data into a database, you're not just transforming it, you're also  copying it.  So you end up with multiple copies of the same data in multiple different storage systems  across your data center.  Doing schema on read means that you have one copy and you can even have the, you know,  you can even project multiple different logical views onto that data as sort of an additional  level of flexibility.  Another core benefit is that since we're building on top of a horizontally scalable  file system and leveraging, you know, the sort of key principle that makes it scalable  data locality, we're able to maintain those same characteristics.  The same goes for fault tolerance.  And then I think the key thing that sort of differentiates this new generation of distributed  databases from Hive is that by replacing MapReduce, we're able to achieve low latency, which means  that you are now able to do interactive queries against small slices of data.  And that, you know, once again really changes the way that you, you know, you can interact  with your data.  We're hiring.  We really like PostgreSQL.  We hack on PostgreSQL a lot.  It seems like this is a pretty good audience, you know, for doing recruiting.  So if any of you folks are interested in perhaps getting a job or changing jobs, you should  definitely talk to us.  That's a link for our website.  We also have a bunch of interesting blog posts.  And also another cool thing is you can actually download a copy of SiteSDB, install it on  your machine and take it out for a test drive.  We have EC2 images available, as well as I think what, CentOS and REL RPMs.  Okay, questions?  Yes?  The license.  So, actually, I mean, so parts of it are open source and parts of it are not open source.  So all of the foreign data wrappers that we've written are open source.  The code that we use for syncing metadata off of the Hadoop name node is open source.  The parts that are closed source right now are basically the distributed query planner  that we wrote.  And I don't think that we, you know, speaking to my coworkers, I don't think we have any  sort of like inherent objection to at some point in the future possibly open sourcing  this.  It's more of just to like, you know, we can't do that right now, right?  That's how I'd explain that.  Yes?  There's a certain threshold of data amounting stuff where kind of usually when you're doing  a closed-grass query versus a kind of additional network of a new power detector and close to  all these various servers, you get kind of like an upper-parabyte.  So I would say like if you're getting close to like, let's say, 10 terabytes of data,  that's probably where this becomes, you know, a practical solution.  Most cases like, I think we have two different types of customers.  So one customer sort of like archetype are people who already have a Hadoop cluster setup.  And they're using Hive, but they're like, man, I can't stand the latency or I really  wish it, you know, provided better support for SQL or I already used Postgres SQL and  now I'm using Hadoop and I really wish I could run, you know, one on top of the other.  Then we have other customers who are, you know, doing things at, let's say like, terabyte  scale, right?  So they have, they're running Postgres SQL, but it's starting to sort of slow down a little  bit in hiccup and they know that they need to go for a distributed solution, but they're  not necessarily interested in setting up their own Hadoop cluster because that adds this  additional management layer.  I didn't mention it in the presentation, but CitusDB actually has two different deployment  modes.  So in one mode, you deploy it on top of Hadoop and HDFS basically manages the storage.  In the other deployment model, every worker node basically runs Postgres SQL, but it runs  on top of the native file system like ext3 or ext2, in which case it just uses Postgres  SQL's storage engine for managing the table shows basically.  We would not actually use Postgres foreign data wrappers.  We would actually create basically like fragment tables on each one of the, on each one of  the workers using basically like the size of these fragment tables is configured.  And then when the user loads data into this sort of like distributed Postgres database,  we sort of in parallel then stream data into these different fragment tables on the worker  nodes.  Yes.  No, it allows.  Push queries.  Right, so it's basically up to the implementer of the foreign data wrapper to support things  like that.  So, you know, you have the ability right now to, let's say, extract filter conditions,  which are something that are easy to push down, but you have to re-express that in terms  of the API of the thing that you're talking to, in terms of the thing that you're building  the wrapper on top of.  So, sometimes it depends on whether you need to change the database.  Change the runtime.  Change the runtime.  Okay, let's take this question offline.  I'm not sure how to answer it right now.  So, the question is if you have a complex query, can you gather all of the data to one  local site?  I'm not sure I understand why.  What problem are you trying to solve by doing that?  Okay, okay, sure.  So for example, the join, right, a join that references multiple tables.  So there are different strategies in distributed databases for doing joins.  So one strategy that we support right now is a broadcast join, where if you're joining  a small table for a large table, you can broadcast, basically, or copy the small table to each  one of the worker nodes so that it has sort of a local copy.  There are other strategies for doing big table joins in a distributed fashion that we're  working on implementing right now.  But the basic issue right when doing a big table join is that you do have to transfer  data between the different worker nodes, because you can't expect that any given worker node  will have all of the data locally stored that it needs in order to process like its fragment  of the join.  And the algorithms that have evolved for handling this distributed join case basically try to  minimize the amount of data that you need to transfer.  So there are different strategies.  In our case, it's a little bit more complicated since we're not able to basically assume anything  about the underlying data in each block.  Since storage is completely delegated to HDFS, we can't do some of the optimizations that  you see in other sort of more traditional MPP databases where they pre-sort or they hash  partition the data across the nodes using a strategy like that.  Makes sense?  Okay.  Question?  Yeah.  So you basically use our version of the copy command.  So like I mentioned earlier, there are two different modes for operating CitusDBN.  So if you're in mode one where you're actually using PostgreSQL's storage layer instead of  using Hadoop, then you would use our distributed copy command.  And you would basically say, here's the file, and I want you to copy it into the following  table.  And it would then, in a distributed fashion, basically stream that data into each one of  the worker nodes.  But in the typical sort of OLAP use case scenario, the data that you want to process  is like fact data or it is log data.  So it's not stuff that you're actually going to manipulate.  You basically want to do joins or aggregations or analytic SQL queries on that information.  And then in the case of deployment mode two where it's running on top of HDFS, you would  use the Hadoop tools for basically loading data in.  So Hadoop has something called distCP which is a distributed copy program, which is typically  used there.  Any other questions?  Yes.  I'm going to actually let Samir answer that because he's been working on this recently. .  Yes.  When Calculator minimum delivery of an incoming fl거h micromoji is nice they have obviously  This version we can for the Hadoop version we can't.  Any other questions in the back?  S och  simil  It seems to me that the  Yes, Krishna?  So we're using that at the master node level in order to replicate the table metadata that  the master node maintains.  So this is basically information that maps a global table name to all of the foreign  tables that exist on each one of the worker nodes.  So it's basically the table metadata, both at sort of the global master level as well  as the map, thanks from the global level to the worker level.  Which is what we want to do, because there's a single master or a single active master,  but you could have streaming replication multiple hot standbys.  And we're not interested in replicating, in a sense, real data, because no real data  exists on the master.  It's all metadata about basically where the shard records are stored on each one of the  worker nodes.  Any other questions?  Okay.  Thank you.
PostgreSQL_on_Hadoop_2_0_1211.png PostgreSQL_on_Hadoop_2_1212_1271.png PostgreSQL_on_Hadoop_2_1272_1319.png PostgreSQL_on_Hadoop_2_1320_1391.png PostgreSQL_on_Hadoop_2_1392_1439.png PostgreSQL_on_Hadoop_2_1440_1499.png PostgreSQL_on_Hadoop_2_1500_1547.png PostgreSQL_on_Hadoop_2_1548_1631.png PostgreSQL_on_Hadoop_2_1632_1703.png PostgreSQL_on_Hadoop_2_1704_1751.png PostgreSQL_on_Hadoop_2_1752_1799.png PostgreSQL_on_Hadoop_2_1800_1847.png PostgreSQL_on_Hadoop_2_1848_1895.png PostgreSQL_on_Hadoop_2_1896_1943.png PostgreSQL_on_Hadoop_2_1944_1991.png PostgreSQL_on_Hadoop_2_1992_2039.png PostgreSQL_on_Hadoop_2_2040_2135.png PostgreSQL_on_Hadoop_2_2136_2183.png PostgreSQL_on_Hadoop_2_2184_2231.png PostgreSQL_on_Hadoop_2_2232_2279.png PostgreSQL_on_Hadoop_2_2280_2339.png PostgreSQL_on_Hadoop_2_2340_2387.png PostgreSQL_on_Hadoop_2_2388_2435.png PostgreSQL_on_Hadoop_2_2436_2483.png PostgreSQL_on_Hadoop_2_2484_2531.png PostgreSQL_on_Hadoop_2_2532_2591.png PostgreSQL_on_Hadoop_2_2592_2663.png PostgreSQL_on_Hadoop_2_2664_2759.png PostgreSQL_on_Hadoop_2_2760_2807.png PostgreSQL_on_Hadoop_2_2808_2855.png PostgreSQL_on_Hadoop_2_2856_2915.png PostgreSQL_on_Hadoop_2_2916_2963.png PostgreSQL_on_Hadoop_2_2964_3011.png PostgreSQL_on_Hadoop_2_3012_3059.png PostgreSQL_on_Hadoop_2_3060_3107.png PostgreSQL_on_Hadoop_2_3108_3167.png PostgreSQL_on_Hadoop_2_3168_3227.png PostgreSQL_on_Hadoop_2_3228_3275.png PostgreSQL_on_Hadoop_2_3276_3335.png PostgreSQL_on_Hadoop_2_3336_3383.png PostgreSQL_on_Hadoop_2_3384_3443.png PostgreSQL_on_Hadoop_2_3444_3491.png PostgreSQL_on_Hadoop_2_3492_3539.png PostgreSQL_on_Hadoop_2_3540_3587.png PostgreSQL_on_Hadoop_2_3588_3635.png PostgreSQL_on_Hadoop_2_3636_3683.png PostgreSQL_on_Hadoop_2_3684_3731.png PostgreSQL_on_Hadoop_2_3732_3779.png PostgreSQL_on_Hadoop_2_3780_3839.png PostgreSQL_on_Hadoop_2_3840_3887.png PostgreSQL_on_Hadoop_2_3888_3959.png PostgreSQL_on_Hadoop_2_3960_4007.png PostgreSQL_on_Hadoop_2_4008_4055.png PostgreSQL_on_Hadoop_2_4056_4103.png PostgreSQL_on_Hadoop_2_4104_9695.png PostgreSQL_on_Hadoop_2_9696_13751.png PostgreSQL_on_Hadoop_2_13752_18863.png PostgreSQL_on_Hadoop_2_18864_22175.png PostgreSQL_on_Hadoop_2_22176_27959.png PostgreSQL_on_Hadoop_2_27960_28043.png PostgreSQL_on_Hadoop_2_28044_28127.png PostgreSQL_on_Hadoop_2_28128_32099.png PostgreSQL_on_Hadoop_2_32100_40067.png PostgreSQL_on_Hadoop_2_40068_40883.png PostgreSQL_on_Hadoop_2_40884_41627.png PostgreSQL_on_Hadoop_2_41628_44039.png PostgreSQL_on_Hadoop_2_44040_44267.png PostgreSQL_on_Hadoop_2_44268_45683.png PostgreSQL_on_Hadoop_2_45684_45923.png PostgreSQL_on_Hadoop_2_45924_45971.png PostgreSQL_on_Hadoop_2_45972_46031.png PostgreSQL_on_Hadoop_2_46032_46079.png PostgreSQL_on_Hadoop_2_46080_46283.png PostgreSQL_on_Hadoop_2_46284_46619.png PostgreSQL_on_Hadoop_2_46620_48647.png PostgreSQL_on_Hadoop_2_48648_50315.png PostgreSQL_on_Hadoop_2_50316_51263.png PostgreSQL_on_Hadoop_2_51264_51407.png PostgreSQL_on_Hadoop_2_51408_51551.png PostgreSQL_on_Hadoop_2_51552_53207.png PostgreSQL_on_Hadoop_2_53208_55139.png PostgreSQL_on_Hadoop_2_55140_55607.png PostgreSQL_on_Hadoop_2_55608_56519.png PostgreSQL_on_Hadoop_2_56520_63707.png PostgreSQL_on_Hadoop_2_63708_64727.png PostgreSQL_on_Hadoop_2_64728_70355.png PostgreSQL_on_Hadoop_2_70356_75707.png PostgreSQL_on_Hadoop_2_75708_78323.png PostgreSQL_on_Hadoop_2_78324_81815.png PostgreSQL_on_Hadoop_2_81816_84839.png PostgreSQL_on_Hadoop_2_84840_104868.png