Apache Cassandra Cassandra Summit Europe 2014, 27 Dec 2014

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: DataStax: Architecting a Big Data Platform with DataStax Enterprise: Cassandra, Spark & In-Memory

Description

Speaker: Piotr Kołaczkowski, Lead Software Engineer (Analytics) at DataStax

You learned everything about Apache Cassandra and want to immediately deploy it in production? Great! But before you do it, don't miss what new features DataStax Enterprise has to offer. The In-Memory storage option allows to significantly decrease latency by serving all reads without ever touching hard-drives, while still offering strong data durability guarantees. And integrated Apache Spark allows to perform complex analyses of your Big-Data stored in Cassandra faster and more conveniently than ever. All of this with easy setup, enterprise-level security, visual management and expert support. This talk will not only cover how to use those new features, but we'll also dive into the architecture and some DSE internals.

A

Good afternoon everybody, my name is piotr kocovsky, I'm a lead software engineer for the dac analytics team in datastax, and today I've got a presentation about some new cool stuff that you'll find in the new destax enterprise release, and also some not so new, but also released in this year. So, as you know, datastax enterprise is the big data platform built around apache cassandra.

A

So it's not just apache cassandra, which we all know that it's a great piece of software for running your transactional workloads, but in data stacks enterprise, you find a complete solution for driving your enterprise online applications. So you get, for example, a built-in search full text search in there, which is also distributed and scalable. Just like cassandra is you get multiple analytics options like hadoop and now the new thing apache spark all of that is with the built-in security like ldp and kerberos options, uh excellent technical support and drivers.

A

So today it's of course not enough time to uh say about all those features in a so short presentation, uh so we selected two things concentrated around basically in memory processing. So how can you do in memory processing with dc? There are two options: if your workout is purely analytics, then you have the new offering of apache spark and if you use transactional workers with pure cassandra dc offers you additional options to make cassandra a full-fledged in-memory database.

A

Let's start with the apache spark project, which is probably the main reason. So many people came here because of the topic. So you know, patches park is a very hot topic right now it is distributed. It is scalable, it is very fast. It offers very low latency of queries compared to hadoop.

A

It offers multiple languages apis, so it has python java and scala api. It has very rich functionality. Well, it's not just mapreduce it's much more than that.

A

It has also streaming it's probably the only open source framework for such distributed analytics, which offers both batch process or batch oriented, processing and stream oriented processing.

A

What's in dac spark, all of the you know that apache spark is apache. Spark is an open source project. So why use dc and do not just use apache cassandra and apache spark in dc. You have all the well batteries included approach, so you just install a dc analytics data center and you have you just select them to run spark and it's ready. You can query your cassandra data immediately, so you file your startup console and it's ready. There is additional python ipa api, which is still not, which is not in the open source.

A

Yet there is also cfs integration. Cfs is a distributed system in apache on top of apache cassandra, which is also dlc specific. There are there's high availability of spark master also built in so you can. If your main master fails, then the backup master can be brought up in seconds and take over your running jobs. So without breaking jobs and to this we have also lots of additional security options, so password authentication, ldp kerberos how it's all architected.

A

uh Basically it's, it's quite a simple solution, so every cassandra node every dc node, will run one instance of spark worker which is kept in a separate jvm. All of those are supervised by one single spark master and optional. Backup reserve spark master for failover.

A

The applications here that you submit to to spark are delivered to the spark master spark master, starts a job and distributes the the jobs to the workers, and then workers starts start executors. So executor is the process that is going to process your data, so every executor would just go to your cassandra. Fetch data from there do some computations and if you want it, also put some data back to cassandra.

A

So let's look how this link between cassandra and executor is done for this? We developed spark cassandra connector. This is the open source project hosted on github, so everyone also can use it with just pure open source, apache, cassandra and spark.

A

It serves as a middle layer between spark executor and cassandra. What it does it not only just transfers data, but also adapts the data model between spark and cassandra, because those data models are a little different, so yeah spark data model spark models, data as collections of objects.

A

So this is not just like tables in cassandra, but you can just imagine just a sequence of elements and elements can be of just arbitrary type. Whatever you wish think java objects or scala objects and those collections are immutable.

A

They're iterable, they are serializable and the main difference between them and, for example, like standard scala collections is that they are distributed. So the programmer, of course, doesn't see it. The programmer, treats it as a simple scala object and works, as with scala collections. So, for example, calls some methods like map filter, reduce every scala programmer knows that it's also by the way in java at similar methods.

A

But here the distinction is that it runs those functions, those methods on on on the on the collection, it runs them on all the nodes and it automatically parallelizes that uh yeah so compared to hadoop.

A

There are many more operators. This is not just map and reduce and well combined. So this is, of course, on this slide. Not not. Those are not all the possibilities. uh Actually, I couldn't fit all the methods that are available on the rdd uh into one slide.

A

So yeah, please refer to the documentation to to read what those methods do, because uh sometimes something that you need to do is just possible to do with single method. Call it's a very rich language. Okay, once you install dc the easiest and fastest way to check how spark is working run, dse spark. This is interactive. Shell, you start dc spark. Your spark context is automatically created for you. It's automatically associated with your cassandra nodes so with the cassandra embedded in in dc nodes and also automatically associated with the spark master.

A

So this whole dc infrastructure knows where those components are living and the only thing you actually should do as a programmer. You write your queries so to access cassandra table. You just call cassandra table method on on the spark context and provide a key space and table name, and you get your table represented as rdd of rows.

A

If you don't like the well programming queries in scala, you can also use your well-known sql language. There are actually two different processors for sql right now supported. One is the new spark sql and another one is kind of middle ground between spark and hive, which is hive ql.

A

So it uses a hive parser query parser, but executes the queries on the spark cluster.

A

You can also very easily deploy your standalone spark application so because, probably okay you're fine with testing it and playing with the interactive shell, but for real applications. You probably want to submit a standalone thing, so it is also quite easy. You just need to bring spark cassandra connector as a list of dependencies of your application, of course, with all the other transitive dependencies. There are, fortunately not very many, then write a standard application with the main method in the main method. You create a spark context.

A

I'll show you in the next slide how to do that package is all in a jar. So it's like a single maven or sbt task and then called dec spark submit.

A

Put your java show tell it what what is what what's your main class of the application and uh eventually put the arguments to your application and dc would automatically register the jar onto spark workers. Add it to the class path and pass it to spark to to execute that code. Let's look how this very simple applications. Tab looks like so just define standard main you can. You must create a spark configuration object.

A

uh You can set some other properties on it like, for example, uh setup amount of memory given to spark or well set up, also alternative spark master or other options. So so, if you, for example, connect from not to dsc but to some standalone apache cassandra cluster, then you would have to put more options, but in dlc it's extremely simple, because it doesn't need any options. It works out of the box.

A

Then you create a new object, spark context, and here you can do all those queries on sc. Finally,.

B

A

Should clean after yourself so call sc stop what what? What is the the api for reading data? I show you that on the first one of the first slides, that is this cassandra table call, but it can do more. Not only you can represent your full table as an rdd in spark, but you can also restrict the amount of data that would be loaded into spark.

A

So, for example, you can restrict the number of columns by calling select and then you can put a list of columns or you can restrict the number of rows by calling where the argument passed to were is actually a part of the cql query.

A

This will get passed to the cql queries produced by our connector and sent to cassandra for execution. So, for example, you can use quite efficiently secondary indexes, even though secondary indexes are not recommended, as jonathan today said, are not recommended for your transactional workload, because it's kind of scatter gather yeah. So it contacts all the nodes, but here in spark the connector would restrict every query going to every node with a token range, so that the query stays within that node and checks just the local index of this.

A

So this is quite performant, of course, as long as your data are not extremely high cardinality values, you can also use it for looking up by primary keys now in the new version of the connector, with just released with 4.6, and you can also use it with uh clustering columns as well.

A

The connector provides not only fetching data in a very generic way like cassandra row so here in this example, this is kind of like with jdbc api or with the native java driver api. You just have a row object and you can call getstring get in to get the the data of the right type. But what? If you? What? If there are more than one possibility to to fetch the data? I mean more types available like with the date yeah.

A

Programmers can't agree which data type to use, so some are better somewhere, not, but it all depends on your context of the application. Where are you using it?

A

So we wanted to make it more flexible and we gave you the generic get method. This is possible thanks to great uh implicit mechanism embedded in scala, so you can pass your type as a parameter and then this way we can support, for example, many types of dates, or we can support many different types of collections and your row objects. Your your columns in in the row object will be converted to the right type.

A

This also allows for a nice scalar idiomatic way of handling null values, as we know null is, is an anti-pattern in scala. You shouldn't use it, it's just for java backwards compatibility and therefore you should be converting those to none or to noun, so so as a subclass of option object. So here you should then request a conversion to optional type to to the type of option and the the connector would do that automatically. For you.

A

There is also some very simple object: bumper, this works with case classes or with the javabean classes. So in the java api, it's also possible to map your tables onto your rows onto instances of java beans.

A

This is convention over configuration, so you should just name your fields in a class in the same way as your fields as your columns in the cassandra table, and it will figure it out. If it doesn't figure it out, then you will get some error message. So you definitely know there are two mappings actually possible.

A

So one convention is that you should be using camel case names on your scala or java side, because it's natural for java and scala programmers, but database administrators are probably not very happy with camel case in the table names because then they have to put those double quotes everywhere and if they forgot, then the columns can't be found. So therefore, it understands the underscore convention on the cassandra side and it does automatic mapping.

A

If sometimes the automatic mapping doesn't work, it always falls back to the safe default, so matching exactly by case, um of course, with besides reading data. You've got also writing data, and this is essential for the solution to be complete.

A

Writing is just a simple. You call safe to cassandra special method, that's added on actually every rdd, so not just the rdd that you fetched from cassandra, so you can construct. For example like here, you can construct in-memory, in-memory, rdd and memory collection by calling parallelize, so parallelized transforms a standard scala collection into a distributed collection known by spark, and then by calling safe to cassandra, you will convert those data in opposite direction, so it will write, write it into the cassandra table and now the most interesting part.

A

This is something that hadoop can't do right now I mean not just not the well the pure hadoop solution. You can store some of your data or even all data of your table as long as it can fit into memory, it can be cached in spark. It will actually reside on the jvm heap of spark. Executors is also an experimental option to put that off heap and accessing such data.

A

Cached already in in memory of spark, is extremely fast. This is this place where apache spark project tells you that they are 100 times faster than hadoop. So you you shouldn't, expect 100 times well performance improvement from just on disk operations. That's not possible, because hadoop is already quite fast at batch processing data that is on the disk. But here, if you cache your data, then you can reach really great performance.

A

The typical use case for that is you pre-aggregate your data from disk to limit its amount, or you filter some of the data and then put into memory, and then you can iterate very fast. uh It has also spark streaming support. This is also new in dc 4.6, uh so spark streaming is actually micro batching. It allows you to accept streaming data from multiple data sources like kafka from hdfs kinesis twitter then divide that into micro, batches and then process every micro batch as a spark rdd.

A

So every micro batch is a spark collection, spark resilience collection. Those collections then, can be saved to cassandra or to cassandra file system, which is like hdfs compatible file system.

A

The dec integration again provides you some additional method like safe to cassandra, which you can call on the stream. This is not in the stock apache spark. This is addition of spark connector.

A

We release also the python api. This is also very new in dec 4.6, the api. We try to make it as similar as possible to the things that are already in java and scala, so the same method name cassandra table. It will, of course convert you, your data to python, I dramatic type. So, for example, it will use python, utf, strings or datetime objects for representing cassandra dates, also python dictionaries for cassandra maps and so on.

A

So all the types in cassandra are supported in all the three main apis of spark: okay, two more words about high availability. How it's achieved spark comes with built in two options for high availability.

A

One of those options is to use a zookeeper and yarn and some quite complex, hadoop stock to to provide higher availability and another is possible to do in the standalone mode, and this is the thing that we used, which requires you to store the spark master state on some kind of a shared volume.

A

But we didn't like the idea of having another shared volume, because it feels like a single point of failure. You you create an nfs or something like this. If it goes down, then you lost spark master state. That would be a bad thing, so we decided to use the something that we already have in dc. This is a cassandra file system which is a distributed, no single point of failure, file system compatible with hdfs.

A

Unfortunately, it turned out that apache spark doesn't offer any connector to store master state in hdfs, so we wrote one and it's included in dse and together with some some detection of node availability provided already by cassandra. Gossip, we came out came up with uh with a higher availability solution for spark, uh okay and but what, if you are not doing analytics you're doing just you want to process some pure transactional workload? There's also something for you in datastax enterprise.

A

This is in memory, cassandra option and from the whole user's perspective, you just need to turn it on and it works as cassandra. Well works. Typically, so you just put data in and you do some queries.

A

You of course, have to remember not to put too much data at once. I mean you have some limits determined by settings.

A

What this thing changes so here on the slide, you see the standard, cassandra storage engine without the in-memory option. So, as you know, when you write data to cassandra, they go into mem table when the main table fills up, it's flashed onto disk in form of ss table, then the next one, the next one, the next one- those if there are enough stables by default. Four, then they get compacted.

A

And when you want to read some data, then they'll read both from the mem table and they also search in some of the ss tables the mechanism for finding which stable would contain. The data is quite complex and quite efficient also, but nevertheless it typically always touches the disk, and this is a pretty significant latency limit. Okay, you can put it on ssds and this, to some degree solves the problem, but it's still not as fast as if it would be in memory so to make it faster.

A

You need to just in dc just add the compaction option, memory only strategy. This would change the cassandra storage engine to something like this, so the mem tables are still being flashed, because this is a standard standard way. How cassandra processes, because it has to flash a mem table to, for example, recycle the comet log and to get rid of some older entries there.

A

But what the compactions the memory only compaction strategy would do, would just take the ss table and create convert it to a new mem table and immediately put those data back into into the mem table by the way. In this process, it also cleans up all the data from expired. Dtls expired tombstones, so after gc grace.

A

So, therefore, even if you will overwrite your data very often this won't grow in size indefinitely. So this is a great, a great way to serve overwrite, mostly workload on cassandra. So if your rights mostly are overrides- and of course your workload is very read heavy and then this is the the best solution for you.

A

Okay, let's see some benchmarks, um we measured latency and particularly the 95th and 99th percentile of latency. You can see a huge improvement here. The the blue bars are actually out of scale here, so so they they were just scrapped by the by the the chart, and you can also see that there's a huge improvement on the average latency, and this is the the greater the more sustainables you actually have.

A

So if your compaction, for example, is lagging behind the standard world, not not in memory cassandra, then you could see you would see a huge difference when converging to the in-memory option.

A

Okay, thank you! That's it generally I'd! I I'd like to answer any.

B

A

The question was about the data loss within memory option. As you can see here, the comet's log is still there, so you can actually have the the durability guarantees.

A

Also it's possible to have the durability here, because if you, if you save your comet log, then you won't lose.

B

A

And the downsides so currently there are some limitations to this, so this will will be better in the next versions. So one of the limitations is that the in-memory option stores your data on the jvm heap right now and in the future versions there would be support for mem tables, stored off-heap as well generally. This would be a generally supported cassandra feature. So then it will be good for really huge heaps and then, if you have a fat node with well 100 or more gigabytes of ram, then you can put put your data there.

A

A

Oh, I'm not sure of that. Actually, the the no the old s tables are dropped in this way, but you can can of course backup them. I mean they. They are not not any more used. I mean the the oldest stable, that was flashed is no more used by the cassandra storage engine because it has all the data in the map table. Of course, if you exceed the size, it will err, it won't accept any more right.

A

So the one of the change that we had to do is to limit the the number the amount of uh data. So it's a there is a component for tracking how much data there is in the mem table.

A

uh Map type, so currently you can fetch only the whole map, so so you, you should avoid a huge maps because they would be just serialized and sent to the spark as a whole. So there's no way for actually you could. Theoretically, at least probably in the future I mean when, when we have cassandra 300, where there is a possibility to query by, I mean query inside collections to index collections, then you can use that work class to to restrict the size of the collection.

A

Also, but currently you can just get a full map and then, of course, on the application side, you can do some filtering processing. This.

A

Oh, I sorry I didn't hear you.

A

The python api is currently only for spark, but I think that you can pretty easily use it still with streaming, because streaming is, is processing by uh micro, batches yeah, so the micro batch is just an rdd. So if you have access to your rdd, then you can call that save to cassandra on the rdd on the python rdd object, but yeah. There is no support currently like we have with scala version where you can call just safely cassandra on a stream.

A

B

A

Okay, I think there are no more questions, so thank you very much.