scale.bythebay.io: Kuassi Mensah, Turning a Relational RDBMS Table into a Spark Datasource
good morning thank you for coming so I'll be presenting this session turning relational database tables into a spark data sources safe harbor this is the legal disclaimer from my company this is me I work for Oracle I do for that management I take care of several products you know integration with Java embedded in the database outside the database Hadoop spark we're looking into flink as well JavaScript embedded in the database or as a client etc see to graduate from Paris French action etc this is the agenda so we'll quickly talk about why are we doing this you know why do you need to turn relational tables into spark data source and then we look at Apache spark I mean if people are familiar with spark I will skip how many people here are familiar with spark and O body ok so I'll skip that part I will go quickly and then we'll go into the substance which is how are we doing that how are we turning relational tables into a spark date sauce I'll give you a demo which will explain some of the concepts and then we wrap up I have 40 minutes laughing I have 39 man okay so requirement why are we doing that we also imagine your VP of Sales or VP of Marketing you know comes to your office and say hey can you tell me which of our product got ratings of 4 on social media in the last quarter ok so to answer this question it could be some other question I'm just making this up also to answer this question you need master data master data product customer you know those are data that you use across applications in your company and there is only one version you know one instance on their usually stored and kept in relational databases but you also need Big Data you know you need the to eat all those things from social media because the question is which product got reading of four or higher on social media in the last quarter so you need Bob data okay so how do you reconcile data start on those different places well you have two approaches you can copy over from our DBMS because this is a smaller data set compared to Big Data so it's a smaller set so usually customers I mean people we know of copy over they copy over from our DBMS to spark so they can do all the join locally okay that's good but the problem is data keeps moving you know the the master data keeps moving and when you copy maybe it's already behind so you need to repeat the operation and I I mean tell you the truth I have met customers they do that every hour so every hour you copy over but that could be hundreds of megabytes or gigabytes of data so that's not really practical and you need to okay so the other approach this is one of the approach I will be presenting is to turn the our DBMS table into spark data source but you don't want to move data it's not like you're gonna suck that transparently through the connector so we want to prove minimize data movement we want to minimize you know data you know preparation whatever and you want to preserve all the security you got from the our DBMS okay so Apache spark I think most of you have familiar so I'm gonna just keep this is a talk I've given a Java one and also at Oracle code online so I need to tell some people what is spark you know some people might not know spec so I need to tell them that okay so I go quickly for them I talk about data frame data source API etc the big picture and you can see at the bottom that there are a provision to access our DBMS from spark data so sepia talked about some data point you can don't care about that how Apache spark works you know a spark start with the the context etc the dark at the try you know this is a very basic example I've taken from the internet which is you know how to go from the HDFS into the Hadoop RDD and then do some transformation you know always transformation and then apply some persistence on the final data you want to preserve this is the workflow you know where you go from the application and then you go from the different transformation I just show you in the back and then you turn those into tasks you know and you schedule the tasks so and then the tasks got dispatched over worker nodes on that's where they have been processed so that's how it works and you're probably for a minute are you also familiar that spark does not come out of the box with the streaming capability the engine is batch oriented it's MicroBot but stills batch so to process streaming data you need spark streaming which is an additional framework that you use on top of your spark cluster this is those are the streaming data concepts you know I ask you know usually when I present people I said there will be a quiz after this slide because I need to ask you what is the timestamp you know what is the event time the processing time the the the the injection time you know and I usually take the Star Wars movies where the the movie the sequence of the movie is different from the history if you look at the story in the book it's different so in order processing out of order processing some frameworks are capable of dealing with that some are not at the Tri cetera so those are some concept that you are probably familiar with but those are streaming data processing concepts okay I'm going now to my presentation this is okay so in order to achieve what we want to achieve we need to look at what's there what is the provision in the spark to allow it to do that so what you have is you have the data source API okay and then you have the data frame API this is how you would produce a data frame from the RDD from the schema in scala and then you have the sequel interpreter an optimizer and the secret service we don't care about the secret service for this purpose okay so so to do so this is the idea you want to access your a DBMS table from spark it means you want to be able to have parallel access to the table so how do you manage to have parallel access to one single table in your DBMS I'm not talking about Charlotte databases this is something we also need to look into you know when you have multiple charts on separate databases but let's say we have one single table how do you manage to get parallel access to one single table but so you need to do some logical partitioning and in the spark framework they provision something called partitioning so partition is allow you to do logic partition of the table for example you have 100 sparknotes so you want to have 100 simultaneous connection to the database accessing that same table so how do you do that I'll explain a little bit how we're doing that so that's something you need to this is important you need to do that okay so the spark itself furnishes a JDBC interface if you look at the data source API there is a DB C interface okay it provides a predicate push down you know the we're close you can push down to the our DBMS remember we don't want to ship all the table on to spa you want to push down the predicate or everything else to the our DBMS and only recoup the result set so it provides some predicate push down and some basic partitioning capabilities so that's what you have out of the box but the problem is it's not enough so all the database vendors you know whether it's Microsoft IBM Oracle or whatever and DBMS you know the no sequel database vendors you know the Cassandra etc they are all coming up with their own spark connector because they go beyond the out-of-the-box solution in spark so they will give you more capability of the partitioning you know how you can slice thing in the most efficient way I'm showing you the class diagram of what we are building you know for the Oracle database and you can see that at the bottom you start with the data frame you know it's instruction from your application and we need to use the default data source and from there we have the Oracle JDBC relation and then we have the Oracle JDBC RDD which is where the actual access of the table is happening that's where all the scans are happening that's where we also do type conversion etc so I'll explain all those later so this is what we're using in on in our implementation and these are all the optimization that we have been injecting into the product so I'll go into each of those the most important are the partition is we have custom partition is I will explain what these are the predicate push down projection push down for example when you select columns you select a subset of all the columns of the table so we need to push down that selection that's called the projection and the predicate the where Clause partition pruning I will explain partition pruning is where so let's say you have hundred access to the table you have logically partition your table let's say in 200 partitions but the query can be satisfied with only one or two partition so you don't want to access the remaining 98 partition you only create two where I mean two tasks and those two tasks will access only two partition that is sufficient to answer the query so that's partition pruning and shout awareness well if you have a Charlotte database you need to be able to direct the worker nodes you know to different shots and type conversion explained connection properties re-explain connection caching will explain authentication you can go from the traditional username password into more strong authentication you know like Kerberos reduce etc encryption etc etc so those are the optimization that we are making into our implementation partitioning or splitter you can see the definition if you go to that doc you know I I put the short link so right now as we speak we have five partition we also call them splitter the reason is we start this whole thing with Hadoop we have provided the mechanism where you can turn or uncle tables into Hadoop data shows and in Hadoop vocabulary they call those splitters and that's why I'm saying splitter or partition um it's a same thing so we have a single splitter which means the whole table is treated as one logical partition so there is no parallelism only one task then we have rows splitter let's say your table have 1 million rows I mean it could be much more than that but let's say 1 million words and you have you have 100 processing nodes so you want to divide 1 million by 100 so that 10,000 rows so each partition will have 10,000 rows that's how you will access the same table with hundred nodes and each of them will be processing one partition so this is a this is very important we also have the block splitter where instead of rows you can use block count it's your table have ten thousand blocks and you have 100 nodes you you can divide by 10 by hunter and then you have hundred 100 by 100 is 10 thousand blocks for example the other is partition splitter where if your table is already partitioned because in our DBMS is the table can be partitioned ok so if the table is already partition we just map the logical partition the logical sparc partition to the table partition so it's one-to-one but it's a it's a possibility that you can use as well also we could extend that notion to chart if you have multiple shards if your database is a Charlotte database and you have multiple chart you could use the partition splitter or some other mechanism to map each worker node to a a chart and then we have the custom splitter the custom splitter mean we give you the possibility to use your own query your own secret statement to do the partitioning the the split because maybe you want to do something we don't keep you out of the box and you have a better way of splitting the logic logical splitting the table with your processing node so we give you this capability of using your own user-defined query okay so okay this is how you will start SPARC shell you know we keep the the jar file different jar file including the JDBC jar file and the software itself you know I put dot dot dot good but it's you can imagine what it is and then you can create your data frame you can see the Val DF you know this is how you will create the data source and you can see here you specify you are specifying the connect string so different connect string you know this is the old traditional Oracle Connect string the or occurred in busy driver that's the name of the table there database table this is the username password but this is for demo purposes I think in real life we recommend that you use more strong authentication such as caporals ssl reduced etc and it's it's not different from this but it's more secure and this is how you specify the partition are you using here we are using the block splitter and you can also see I don't want more than false partition enough because the splitter might give you one thousand split you know if the algorithm is not well defined you can cut this on say maximum number of split or partition is this number in this example we say maximum partition is four but it could be more than that okay and then you can you can load you can create a data frame this is the definition and this is where you actually create the data frame but there is no rows fetch right now this is where drawers are first and you can implement some filter this is the where clause so this is the were close and then you can say show me the results so that's how you could use this with this spark shell okay so other optimization or one thing I did not mention when I say split and we divide logically by 100 we rewrite the query for each split you know the software will rewrite your query because the database has no idea this is coming from spark and if you want one task to access only one logical split you need to rewrite the query and say ok you execute this query where row ID from 1 to 10,000 on the next guy you need where row ID from 10,000 1 to 20,000 etc etc so under the cover the software rewrites the query for each of the partition and then type conversion this is important because we are going from sequel types to Java types because we essentially DC to access the database so we're going from a secret I'll treat other types to spark type ok that's double conversion so to avoid that in the driver we have provisions of mechanism where we can go straight from secret type to spark without the intermediate combustion you know so some you know we can provide you some raw byte we can provide you the representation the byte representation of date/time etc etc ok a connection property all the JDBC properties are allowed for example you can specify the fridge size you know they have to do Arafat you know you don't want to get rows one by one so you can specify how much world how many rows do you want for fare for each face you know the resulting connection cash means so I have I have generated hundreds partition but I don't have hundred worker nodes I only have fifty which means I have to exit I have to have to do this in two steps the first step I will use 50 tasks and they will execute the 50 partition and then I will read I will use the same 50 workers with the the remaining set of the the partitions so in order for that we don't want to recreate the Carnation you know in order to do that so we catch the connection it's not connection pooling it's connection caching within each worker so the worker will reuse that same connection with the same authentication and to process the remaining partition but at the end of the query it's over so for each query you have to have a different authentication because that's a security reason because you might be accessing a different table so we don't keep that we don't keep the connection when the query is over but until the queries over we have the possibility of caching to reuse within the same worker and the authentication so you have all the publication that JDBC allows you you know the SSL Kerberos reviews wallet jks etc cetera a che you know all the database a checkup abilities that the JDBC driver gives you you get those for free through the connector wrong time balancing of requests you know let's say you have a cluster of database and you know that some nodes are overloaded some node are not very loaded so you want to you want to balance the the request the credential request across or the different nodes you know all the nodes are accessing the same table but the connection to the Noda unless you can kill it so you you see what I mean okay time for demo through the demo I think you will better understand everything I just say I hope so I have a virtual box here okay so the virtual box it's it's called the big data light you know so big data is the Oracle Big Data cluster it has both Hadoop spark etc to help people play with we have this virtual box called BD light and it's a vitter box you can download from the Oracle technology Network and it has everything it has the database it has the spark they had do all sort of software so that's what I'm using to show you the demo okay so the demo is a bunch of script I'll have the I'll create I'll show you how to create the Iraq or table how it is created I'll show you how we do a predicate push down how do you specify note that you want to do how will see how predicates push down partition non pruning and partition pruning and join so I'll show you all those and so the way I have this shell script and I have this Calif ID for example okay let me show you the table first so this script will create the table and you can see here that we are using this file which is the sequel which is a sequence file so let me show you what this is data okay so this script will create the table so we're dropping the table and we're creating the table and you see this is a partition table and it has five partition and the partition key is the salary so every salary less than sixty sixty thousand go into this partition and that attack so you see a five partition depending on your salary and we also have something called employee bonus report this is where we will insert the bonus from the employees we will use this when we do the join okay so and then I have okay so line now let's go into the some of the demo that has 15 minutes okay that's good view predicates predication post and that s it so you can see here this file is in working spark shell you know with all the jar file different jar file but more importantly it's using this Scalla file ok so let's look at what is in this scarifier in this kind of file you can see here we are creating this data frame okay with all the specification the URL the URL to connect the driver the name of the employee table the username password blah blah blah and then we we we are doing a transformation you know we're doing a dataframe filter and we're we're adding the query which is the Select blah blah blah blah blah and then I also add explain to show you the explained plant so you can see what is exactly being done okay so if I execute predicate dot SH it will invoke the spark shell and we'll use the skeleton I just show you okay so this is the demo and everything is on my laptop you know so the virtual box everything is here so don't look at this as an indication of speed okay so what's going here and ER the cover here is that I do not show you the specification of the splitter okay I will have to look into that later but it will show you which is the split that has been used so what is happening now is the software you know our connector will take the query will get an execution plan from the SCADA query optimizer and will introspect the table and based on the partition key that it's been given you know and the partition sorry the partition is single splitter you know this is the partition is a single splitter which means there will be only one split generated okay because we say single split so Mac split equal one only one split will be generated and you can see number of split generated this is here it's one okay and we can see that the predicate has been pushed down if you look at the plan you can see that the pretty kilt was pushed down on the wear clothes was pushed down so the results said that was retrieved from the database is this one in other world the database process the query and returned only one single row okay so spark just received that word it does not do did not do any work let me look at some other so fact 3d a partition partition non pruning say partition on pruning SH you know it's the same thing I'll show you this camera because it invoked this kind of fire so I'll just show you this kind of fight okay oh it's easy yes sorry okay so creating the data frame and this is the partition splitter here we specify partition splitter so remember the table has five partition right and we it's the same query as the previous one it's the same select employer blah blah blah with a we're close okay so let's execute and see what happens okay so I come here okay partition on pruning s8 okay that's good okay so we will look at how many partition have been generated and whether or not the predicate has been pushed down or not those are the two thing we will look at so the software connected the table then based on the partition splitter that it was given it will generate a number of partition and rewrite the query for each of them so it's not the dump connector it hits a smart it can recreate the query the Oracle database query and the same tax might be different from this back sequel statement okay so it's not the same dialect okay so what we're seeing here is that the partition the type it recognizes that is okay and then how many partition has been generated is number of splits generated five so all five tasks access the database and each of them was processing its own partition and then we retrieved the result set okay now I'll show you how we do partition pruning okay partition pruning we will view partition pruning skaara okay same thing here we specify the partition split username password everything else but the query is different this query so you see the workload is based on salary salary higher than 72,000 and lower than 78,000 and then we'll explain the plot okay so let me say okay so in principle it should only generate one split because only one partition has this salary range so it does not need to access the other partitions and that's a partition pruning okay so it connect it takes the skylark I mean it takes what was it was given as options and the the plan that it's gets from the sky the spark spark query optimizer and that will generate the number of partition rewrite the query for each of them and that we will see that in this case only one partition only one split has been generated now you can see here one partition and then the predicate has been pushed down so it pushes down the where clause which is the salary right okay okay that's good and last thing I want to show you we have eight minutes I can take some question after that so view join that is Sh so here we will use the join that scarra okay let's look at the join that scanner okay so we create the data frame from the employee data we use the block splitter okay and then we register that we register that data frame you know as employee data Oracle and now we say like we have another query here another query here okay which is joining you know that data with a local data I local spark data okay to produce the the result okay so the where clause is employee bonus the bonus is higher than 7000 and the salary is higher than 70,000 only those people will receive a Polish the this free bonus is on spark okay so we're joining data from spark and data in our DBMS which is the the use case I I start right talked with okay so let's execute it okay this is a two step so it's gonna take a little bit more time it will first execute the first query registered the first data frame register and then to the second one and it will generate for for splits to do that okay so while this is happening I think I can take question we have seven minutes left I'm happy to take a few questions the software is not yet released it on the development and every fortunately it's almost done because everything I show you is based on alpha a better version of the software the thing we're still looking at is the sobbing shining support and some other stuff in minor stuff but frankly it's almost done so we we will do some stress testing performance testing etc cetera okay I'm happy to take questions and you can see the Mac split and I need we do the second step and if we push down the predicate problem Rindy will join so that the joint is happening at the spec level you know we are not pushing down join today so spark will receive the rose from Oracle and will join with local data if you are accessing to Oracle tables and you want to join them you can create a view the view will be the joint so we are not currently supporting join in our software but we're looking into that so we will at some point be able to push down join into the our DBMS but you can use the view to work around that if you will execute the journal Oracle and you will only express your query on spa with the view the name of the view okay any question yes oh you can persist you I mean the data frame you can persist them and then do it in two steps yeah you could persist but the idea is you only retrieve a small set of rows from the database to spa so you so you're talking about persisting the rows when we are retrieving from the database yeah but boy we we don't want to select all rows you want to only you want to execute the query against the tables and only retrieve a small set of rows so sparkly which receives is small set the result set now we don't want to ship data around that because that's that will be equivalent to what people are doing today well every hour they ship the whole table to the spa yeah so we send as the we sent a query we rewrite and send to spa a tooth or alcohol and only retrieve the result set so if you have 100 100 simultaneous connection you will retrieve data from some of them and then you do the the spar query coordinator is doing the final you know you know consolidation because he is the master he owns the whole plant we only receive a portion a portion of the plant of the execution plan okay the question is can we be able to write back to the a DBMS yeah so we know we have not yet looked into those but I think it should be possible once you have the data frame everything can be yes so we extend the the data source there's out-of-the-box data source with our own classes you know threads grab habla and we have not look into the pipeline and everything you ask so no yet no yeah thank you we have two minutes yes yeah we can't persist data back to the our DBMS if we want to I mean we provide that possibility for the Hadoop and we're looking into that for spark as well yes oh that's for the demo I mean this is alpha code no no we will support the spark 2 dot X and we will maybe only support 2 dot X no one point something okay thank you very much for coming are we good [Applause] you