The Spark of Big Data: An Introduction to Apache Spark

Speaker 1 [00:05]

Thank you very much. As I already talked, it should be mostly a live coding session, like we have maximum 10 minutes of slides and then live coding and this live coding we will do without an actual IDE because my machine is terribly broken, I can't install anything, I won't have, even after completion, and I won't have PySpark installed on my machine. So we will improvise the whole stuff, but first of all, yeah, who am I? Yeah, if you don't see me from there, from the from the rows far away from me, I look like this. I actually not look like this, I like this photo more than actual me. Photoshop makes absolute magic to me. I'm a developer at JetBrains. Usually we don't fail our live codings like this, but I did it. I somehow managed to completely break my machine. I'm an average Python enjoyer. I mean, I'm not so long with Python. maybe four years or something like that, I can write some code in Python. Maybe even in Django, maybe not, who knows. Definitely, I am good in Python for data engineering. I can definitely write the code for Airflow if you know what it is. And if you don't, it doesn't actually matter. Also, I am a speaker and streamer at the same time. I speak at conferences, I speak in podcasts, and I stream things on Twitch and on YouTube. You can find our Twitch by this 3moji URL, twitch.tv slash jetbrains. Yeah, actually, my friend, he's probably not here today, but we are streaming how he's a real Pythonist. I mean, he writes code in Python for many, many years. And now he learns how to write a code in Kotlin, and it's kind of funny to watch. And I'm trying to learn from him how to actually create games. And also, I'm a data engineer, and it is what leads me to this talk and to talk about data, because, you know, people sometimes think that data engineering is something hard, it's something difficult to understand, and maybe not even very useful for you. Like, who are data engineers here? Nice. And who are back-end engineers and data scientists? Okay. Awesome. So, folks who are data engineers probably won't hear many new things today. Probably they will, but I hope not. I think that you are already working with PySpark at some level. But a lot of them, yeah, I think that you don't need data engineering skills. It's just a click bait. You can't actually click it, but it's just a click bait. Because you don't need data engineering skills until you do. I mean, what are actual data engineering skills? Well, that's data modeling, like how should we put our data in our warehouse? And actually, warehouse can be your small postgres or even .db database. It's already data modeling. Data architecture, nobody knows what the pipelines of this stuff. large amounts of data, well, nobody knows what is large amount of data. I don't know what is large amount of data, actually. Sometimes I say that if you work on like very old machine, one gigabyte is a huge amount of data for you, and basically it's true, and you don't need the skill because you usually don't work with such a big amount of data. JVM, yeah, Lots of data engineering tools are built on top of just something JVM. Even Airflow, you write the whole code for Airflow in Python, at least I do, but the JDBC connectors are like Java ones, right? It's crazy that pure Python technology is built on top of JVM. Of course, PySpark is built on top of JVM, too. Spark is written in awful Scala. I can't even say how awful it is, but it works. I mean, kind of nice. Debugging. I know some people, all kinds of debugging actually. I mean, print, debugging prints are counted as debugging too. I usually sell this point for PyCharm that we have step-by-step debugging given on remote machines, but probably you don't need it. I don't need it usually. I just set up one print, and it's not there. Something is broken before. Nice. SQL. Okay. I believe that everybody works with SQL here. Who doesn't work with SQL? OK, one person. Thank you. Thank you for being honest. It will be hard talk for both of us. By the way, you can see the link at the bottom left. It's actually my link tree for you to be able to get all my contacts sent. And give me feedback how awful my presentation was. Did you notice that I didn't mention the actual big data? I don't like buzzwords, first of all, but also it's completely irrelevant to my talk because PySpark is not about handling big data. Yeah, you can handle big data with it, but it's not the point. Even distributed things, distributed computation is not about, is not what my talk is about. My talk is about, I don't know, usage of PySpark for mere mortals. By the way, all the images in this presentation are generated with Midjourney. Thanks, Midjourney, for version 5. I mean, photos are much more nice. PySpark allows us to work with big data, yeah. But it allows us to work with big data, like it's small data, but it's a seamless distributed manner. Like, you usually don't care about how is the thing working in a distributed way. You don't think in your code in a distributed way. You just do your usual function calls, and it does the trick. But yeah, it has only one basically new idea in it compared to usual Python development. It's a concept of data frame. Data frame is basically a distributed collection of data organized into named columns. I didn't even understand what does it mean. I mean, it's a very complicated way to say that we have our data distributed all over the place. I mean, several machines, several data nodes, several data centers, nobody cares. Yeah, what's important, it's immutable. We cannot change data inside the data frame. Like, if it is there, it is there. We can remove data, like remove columns or remove rows from it. We can add more columns. We can somehow add more rows to it. We can filter group join stuff. Like, everything we can do in SQL usually, usual data conveyors, I'm not teaching you this stuff. And yeah, thanks to ChatGPT, this is the basic example of data frame at the bottom of the slide. Like, four columns, ID, name, age, and salary. And what we can do on this data, I don't know, we can add one new column, like twice a salary, and for John Doe, it will be 100,000. Congrats to John Doe on having a huge salary. Yeah, he's 32 and probably white male, I don't know. So yeah, but what we can't do, we can't change his actual salary of John Doe. We can't say inside our data frame that now your salary is 80,000. We should create a new row, a new column, we will modify data somehow, and then we should drop the old salary. And, yeah, it was a little lie that it's only about data frame, because, yeah, actually Spark gives us a distributed computation engine. It means that basically we can solve any computational task in Spark, any distributed computational task. Or not distributed, if it doesn't distribute, Fibonacci numbers are perfectly possible to do in Spark. I don't know why would you do so, but it's possible. Or if you need to traverse some huge graph, I don't know, insane graph, virtual, you can do it in Spark. And this is what makes it so popular, as well as it can connect to multiple data sources. You usually don't think about it, but many of you work with relational or non-relational databases. And then you grab some data from this relational database, and you make some HTTP call to grab some more data, you enrich one data with another data, and now you have big ball of data which you should somehow process in your own code, right, this for loops or iter tools. By the way, who use iter tools in their practice? Yeah, that's what I thought, less than half. I actually don't like iter tools because I should put one call of iter tools inside another call of iter tools. I like chains. Yeah, I came kind of from Java world. It's fine there. But Gwida doesn't like call chains in your actual code. But sometimes it's just handy to not have many list comprehensions, but to have one thing after another. Then Spark can, I don't know, join data as you like, build execution plan for you, and execute it in hopefully most optimal way. It's a way to say that it tries it best optimize what you are doing, but if you fail terribly, it will fail for you. Absolutely terribly, even worse than me. Why do you need it? Well, first example I already gave you. One day you will decide that you need to join your data from your actual database with some data from HTTP API, and it's fully achievable with Spark. Maybe collect data from multiple sources. I don't know, once upon a time, I've been working at some big bank, and they had lots of branches, and they had to collect data from all these branches, and somehow handle it all. I wrote some terrible thing, I had a for loop, and HTTP call inside this for loop that will put each JSON into some list, and then I try to handle the whole list. It's actually not the best way to do it. With Spark, it would be much easier. Or maybe sometimes SQL is not enough for you because you need some function that is not available in your dialect of SQL. You know, last days, these days, I usually talk about DuckDB. DuckDB is an awesome SQLite for analytical purposes. Like, it works on your machine, it has lots of analytical functions, I believe more than 100, almost as Exadata. But if you use MySQL, it happens to the best of us. There are like 20 analytical functions, it's not enough. In PySpark or in any Spark, You can write your own function that will aggregate data as you want it to be aggregated. And yes, of course, it can handle big data. And by the way, on the picture, you can see a spark ignites big data. I couldn't even imagine that it's the actual illustration, but yeah. So our today's sample, and by the way, we just finished the best part of our presentation that actually doesn't fail. Today's sample is MovieLens. MovieLens is a recommender service that allows you to find a movie that you will probably like based on your taste. You set some mark to each movie you watched, and they try to recommend you the next movie you should watch. Works not Not so bad, but in their official data set, they have data only until 2018, and so yeah, it's not completely up to date. But still, I have a question. How do you think, what is the most popular genre of movies? Who thinks it's comedy? Two people, three people, four. Okay, folks, let's say from 20th year, like 1920th, to 2018, the last, only last hundred years. Yeah, so yeah, action was like eight people, drama, okay, like nine people, ten, western, One person, okay. What else do we have? Film noir. Yeah. I didn't even hear about it before I started handling this data set. Actually yeah, it's a funny story. It appeared like in the 40th year and disappeared in 46. Maybe it's, I don't know, connected to war or something. What else? Which are the most, yeah, thrillers. Okay. The horror movies? Okay, so drama and action are your most popular choices. I have remembered it. Again, I won't be able to show you the actual demo, so we will improvise. I already have, I had like five minutes to download the data set. It's called ML Latest. It looks like this. I don't know if you can see anything. Do you folks know how to enlarge anything in this? Yeah, like this, okay, nice. At least something Mac can do. So it looks like this. It's just CSV with three columns, movie ID, title, and genres. Yeah, ID is just a number. Title actually includes the actual title and year when it's done. And generous is just a pipe-delimited thing. It's not sorted by time. It's not sorted by anything, actually. We also have several more tables in our data sets. What can be interesting for us, for example, tags. I'm not very good with mugs. Give me a second. Tags look like this. Manager ID, movie ID, tag, and timestamp. Yeah, I will have another question about text a little bit later if you will have enough time because, again, presentation goes terribly wrong. Let's say that we want to answer our question of what is the most popular thing in, yeah, I should probably go to presentation mode, at least. Come on. Give me a second. Yeah, enter presentation mode. Yeah, like this. I won't even be able to show you inputs. That's funny. Yeah, so what we are starting from is we should create a Spark session. It's the thing that allows you to actually work with Spark inside some Spark context. They have a field builder and we will call get to create. I wrote it so many times that I memorize it. And of course, yeah, wrong, wrong, yeah. And it will be called something like Spark equals, yeah, something like this. Then, when we have the spark, we want to load our CSV. In our example, it will be CSV. In your actual examples, it could be connection to database or something like that. It will go like this. Spark.read. It's actually a spark reader thing that allows you to read from something. Then CSV. Then we will have to put our path to CSV here. will be something like movies.csv. I don't have an actual code, so I don't have to put an actual pass here. And also we will pass a couple of parameters there, like header equals true, because we have header in our CSV files. And also in first schema equals true. In first schema is kind of interesting thing. can try to understand what are the types of columns we use in our CSV by their content. And obviously, for movie ID, it will be integer or something like this. For other things, it will be something different. Okay. Now we have and we put all the stuff into something called df, it's just a data frame which we read from movies. What we should do next, let's try to open our movies again. What I want to do, I want to split all these genres by pipe because I need to count how many entrances of each genre there was every year. So it will go something like, let's put it in parents like the f dot with column. This way we will add a column to our data frame that will contain a new content for us. And our new content actually is just split genres. For that, this column will be called, I don't know, genrs array, and we will put the split of column genrs by backslash pipe, yeah, something like this. After this operation, and I'm not able to show it to you because my idea doesn't work, we will have a new column, array that will handle the whole list of genres instead of one big string. And let's call this with genres. It's still a data frame with one new column. But by the way, probably we need to do something else. We don't need our genres column anymore, right? So we can drop it. It will go like this, drop, generous, because we don't actually care about it. Now the next task. Our next task is when we have array, we should convert this array to several rows, right? Toy Story is action and comedy and animation and something. We should have Toy Story action, Toy Story comedy, Toy Story animation, and so on. For this we have very nice function called explode in spark. So it will be like this with Jenner equals with Jenner's dot with column. Oh no, sorry. Yeah, with column, right. With column, it will be called Jenner and here it will be explode. and we will use function called col. Col stands for column in PySpark, generous ray. At this point, we will have a structure like, give me a second, this. One, Toy Story. animation to Toy Story, I don't know, comedy, and so on. I'm working for Spark, does it work? Yeah, nice. I'm here instead of Spark for you, like interpreting the code like we do on our interviews. I hate coding interviews, and now it's the worst coding interview in my life. I mean, usually I have like two or three interviewers max. Right now, here are like 30 interviewers. I hope that you don't know PySpark and you can't find any mistakes there. Also, I hope that you don't know Python at this point because I feel that I'm terrible in Python right now. Yeah, so we actually have this, I don't know, Toy Story was, I believe, of 1984, let's check. No, it's 1995, so I'm not that old. 1995, now we should put a year into a separate column, right? It will be something very simple, something like with year equals with gender, dot with column, stupid PyCharm, you could already understand that I have this function. I mean, I'm using it for the third time. With column, yeah, and surprisingly, we have something like regExpExtract. From which column? From, obviously, title. What should be extra? It should be r. What will be the actual regexp here? I believe it will be something like backslash s plus then backslash... Who can read regular expressions? Oh, so you don't actually care. So I will put some random stuff there. I mean, for real. Actually, I'm just looking for these four sequential numbers at the end of the line. Probably there are some more spaces. and the most important, end of the line, dollar sign. I mean, they should pay us at the end. Yeah, and it should be first group of match. And at this moment, we will have almost the same stuff. Please select it for me. Thank you. Thank you. But with additional thing called like this. Also, if you will look at our data set, we will probably, what the kind of name of movie is it? I don't know. I mean, how should I extract it? Yeah, but we will probably find something like, yeah, something like this. No genres listed. We don't care about these lines, right, because they don't benefit our task. We should remove anything with no genres listed. We should remove anything with year equals null. It means that my regular expression is not so good to extract year in every possible situation. So we will take our with, by the way, with year, and thank you. Filter. I actually like how it looks like. Filter. Column. Yeah. Dot is not null. Yeah. I believe it's CamelCase, But I'm not sure. And column genre. Not equals. No genres listed. In parents. Welcome to my stream, by the way. It's how my streams work all the time. But usually my ID works. At this point, we have kind of clean data set. It's what data engineers usually do. They should clean their data, they should validate it, and say, okay, now we can actually work with it. So let's call it clean. I'm not even sure, yeah, of course, it should be not double, it should be like this. Yeah, nice. Now we should aggregate it somehow, right? we should find count of movies in each genre in each year. It will be something like clean.groupBy. We need a tuple here, I think. First column is year. Second column is genre, logically. And we need to aggregate it. And it will be actually very simple. It will be something like call title.count. And we will have something like, I don't know, 1984, comedy 1. and everything like this. Yeah, there were actually more comedies. So if we could actually launch this code, if it wouldn't be just a text editor right now, we would see that the actual, actually the most popular genre all over the time is drama. So those of you who guess drama, you can applause to yourself, it was a good job. Yeah, you definitely know something about movies. I didn't know. I thought it would be comedy. But yeah, I mean, it is what it is, right? We still have some time. So the next important, I promised you that we, just imagine that we have executed this code and it actually works for us. But now we wanna understand a different thing. We have text for movies, right? Sometimes we have many text for movies. How do you think, which is the movie with the biggest number of tags in it? Like, I'll give you a hint, there are more than 10,000 tags for this movie. It's all user-provided content. Any ideas? Do you know something about trackers, or I don't know, other people who enjoy movies? Okay, seemingly no idea, cool. It will be a surprise, I like it. So now we will do basically the same thing. Now we don't need to create a new session. We will read the new dataset again. Like this. And we will call it text. And the code will be very simple. We have already this clean thing. We will join it. Clean.join tags. Does it remind you of anything? It's kinda SQL-like thing, right? Yeah, it's actually, for me, it's the most powerful part of Spark. We can join everything with everything because everything is just data frames. We should define predicate there, and predicate will be actually very simple, movie ID, because they share this column, movie ID. Tags reference movies by movie ID. And yeah, it should be inner join. It will go like this. Then we need to aggregate, what do we need? count tags for each movie. So we will group by movie ID and aggregate by column. It's something like array aggregate. I don't remember the actual syntax, so we'll just emulate it. Array aggregate. Tag. Let's say it's called tag. And we will have something like, I don't know, something like this, movie ID one, tags will be, I don't know, A, B, C, D. Now we need to add one column with tagCount with the length of our tag. And this way it will be converted to 4. We will have this, and also form. And this is an actual length of our, actual number of our text for each movie. That simple. I mean, I don't even know which Python code should I write to join things. I mean, while we were in this, we are in single CSV mode, it was kind of simple, right? Just couple of for comprehension instead of this explode function, and basically count gener by year, right? But when we are talking about joining sets, it becomes not so simple. We should put something in dicts. What should be key in our dict is not very clear, probably ID. How should we find data in both dicts, and how much code we should write? It's a little bit obscure, I mean it's like 10 lines of code, and here I need just one manipulation of columns and that's basically all. And yeah, the movie with the most text is Star Wars number four, like first movie in the series chronologically, and it has 26 something texts, thousands of texts. Some of them which are incredibly crazy. They actually describe the whole plot of the movie in one tag. I wanted to show you in the actual IDE and how it looks, but no luck here. So yeah, sometimes for me, and actually this is the case, you can not only count things. Let's say we have counted all the stuff and let's say it's ready. Now we need to write it somewhere to our database or to our other CSV file. It will go like this. Ready.write.csv and .csv. Yeah, I'm very careful about my extensions when nothing works. And yeah, it's done, you can run it every day because your data in your database is actually updated, right? You should run it probably several times a day. And it's easy, you can run it locally, you can run it inside your actual backend application, you can run it inside, I don't know, some business process, just create one instance of Spark session and run everything in it. When you will go large, of course you can create a cluster of Spark, you can submit your jobs there, and so on, but it is already useful when you don't want to mess with loops and ifs and so on, when you want a kind of declarative approach to your data, right? So yeah, I think it's time to get back to my presentation because it's time for a small summary. I wanted to say that we saw some nice plots, but we didn't saw them. So only my awful handwriting. But PySpark allows us to do several interesting things. It allows us to join data from different sources. In our case, it was two TSVs, already not absolutely trivial in Python, but we can join data from, I don't know, Snowflake, Oracle, and Parquet files at the same time. And for Spark, it's all the same. We can process this data in parallel. It's important when we have large amounts of data, we will launch our stuff in the cloud, it will be massively parallelized on hundreds of machines, and it will run fast, and for you code will remain the same, the same configuration will be different, code will remain the same. And we can work with data in a uniform way. Thank you very much for being patient, for those who was patient, and I think it's time Time for your questions.

Speaker 2 [36:55]

All right, thank you very much for the talk. There's been lots of questions. The first one, I think, that comes from all of our hearts is you've passed the coding interview with flying colors. Thanks for doing this. All right, so Christian asked the questions What's your recommended way of setting up a cluster and submitting jobs with all dependencies like Python packages and database drivers? Yeah

Speaker 1 [37:19]

Yeah, when we are speaking about Python, I would say that we have only one way. When we have Java, we can pack everything in a fat jar and submit it. But otherwise, we just should put our main Python file and all satellite Python files, if you have many, and submit them, and put your jar files as one fat jar, if it's possible. Otherwise, as single jars, or even as dependencies if you have local artifacts, I would not recommend to depend on external repositories. And I would not recommend to depend on libraries we already have in our cluster. To me, it looks a bit fragile. I had issues in my experience with it.

Speaker 2 [38:08]

Nidia asks, what would be your tips to improve writing performance in Spark?

Speaker 1 [38:14]

Writing performance. Thank you very much. It's a very interesting question and kind of deep. I would say that it strongly depends on our hardware. If we work with something not parallelizable, we should probably repartition our data like in, I don't know, two or three big partitions and write them slowly but surely. If we have something like S3, we can do many, many partitions and write them at the same time. But everything depends on our hardware, on our network, drives and so on. There is no, I don't know, single working recipe.

Speaker 2 [38:53]

Makes a lot of sense anonymous asked would you be able to provide this as a working example later and if so where much

Speaker 1 [39:00]

much appreciated for sure it will be on my github my github is asmodei uh please find me at my booth at jet brains and we will try to arrange something i will try to do it tomorrow i wanted to do it today actually i wanted to put it on github today but my machine doesn't work probably i won't be able to fix it today so hopefully tomorrow uh please reach me out in any in any way that works for you and I will give you a link.

Speaker 2 [39:30]

Cool. Another anonymous question is, why did we use Spark for this example? Wouldn't simple pandas or an Airflow DAG be enough?

Speaker 1 [39:37]

Thank you for this question. I actually don't like Pandas for several reasons. I mean,

Speaker 2 [39:38]

Thank you for...

Speaker 1 [39:44]

Pandas is awesome. Everybody knows Pandas is awesome. But I find its syntax confusing sometimes. I'm not sure that everybody can, like, the Navais user can distinguish between series and the whole data frame. But why do we have two concepts when we can have one? Lots of operators have overloaded, it's not absolutely clear to a novice user that we can rewrite column, just assigning it like in a setter. Of course you can learn it on each courses, but the very fact that you should learn it, I don't like it. And also pandas are local. SpySpark gives you this flexibility that now you can work locally and tomorrow you will will go distributed and you don't need to do anything with it but some minor changes to your configuration or not so minor if you write something complex.

Speaker 2 [40:43]

Another anonymous question, how do you use custom functions in Spark?

Speaker 1 [40:49]

Okay, custom functions are called UDFs, user-defined functions, probably the answer is when we are in Python we don't, but we can. Why do I say that we don't use them in Python? Because there is a cost of call UDF from Java to Python. It actually pipes the whole data, every row of data through pipe into the Python process, And it's slow, but still it's totally possible to define UDFs. I would propose if you need the custom functions, if you like something, to write it in Java or to use some Java library. It's not that complex. You should implement one interface with one method and actually Java is very simple language. Just do your stuff and then cost will be like almost zero, call of cost for this function. Of course, it won't be optimized as well as Spark optimizes its built-in functions. But for other stuff, it will work awesome.

Speaker 2 [41:56]

Thank you. Another anonymous question. You say we can run the code whenever we need. What if we want to automate? What tools can you use?

Speaker 1 [42:04]

Any orchestrator, my weapon of choice is Airflow, some people prefer Duxter or Prefect. And of course, good old cron will work too, why not? I mean, if you don't need this enterprise solutions, you need to do something on one machine, just set up a cron there, say to run every five minutes your Python script, and it will do it.

Speaker 2 [42:30]

Another question, do you have a link to a repository with the movies examples?

Speaker 1 [42:36]

It is called ML Latest. Of course, I have a link, but again, everything is broken. Let me try to find it in Safari. It goes like this. Movie Lens Datasets. By this title, you can find basically everything you need. And there are some more interesting examples. For example, Wikilens that has some statistics over Wikipedia and many other things.

Speaker 2 [43:06]

All right, and final question. Is there a difference between using Spark with Python versus Scala?

Speaker 1 [43:12]

Yes, but we have two major differences. First of all, Spark is written in Scala itself, so you can expect a little bit better performance in Scala, but nothing dramatic. I mean, most probably you won't notice it. But there is a more important and more interesting aspect. In Scala or in Java, we can use typed data frames. Like, as always in Java, smart compiler can check types for us. If your column is integer, it won't allow you to use it as, I don't know, date, for example, or vice versa. You won't be able to use date as integer and call an incorrect function on it. In Python, it's possible, but of course you will catch your error with the first test you will write. It's not that big of an issue. Still, I prefer type things when it's possible or irrelevant.

Speaker 2 [44:13]

All right, that's it with the question. Let's thank the speaker again.

Speaker 1 [44:16]

Thank you, Michael.