[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15400394#comment-15400394 ] Stefania commented on CASSANDRA-9259: - Now that CASSANDRA-11521 is ready for review, I've repeated the Spark benchmark defined by CASSANDRA-11542 using schema 1: {code} CREATE TABLE ks.schema1 (id TEXT, timestamp BIGINT, val1 INT, val2 INT, val3 INT, val4 INT, val5 INT, val6 INT, val7 INT, val8 INT, val9 INT, val10 INT, PRIMARY KEY (id, timestamp)) {code} and schema 3: {code} CREATE TABLE ks.schema3 (id TEXT, timestamp BIGINT, data TEXT, PRIMARY KEY (id, timestamp)) {code} The benchmark measures how many seconds it takes to count rows and to find the maximum of two columns for each row, where rows are retrieved either via Spark RDDs or Data Frames (DFs). The most significant difference between RDD and DF tests is that in the DF tests only the two columns of interest to the Spark query are retrieved, whilst in the RDD tests the entire data set is retrieved. The data is either stored in Cassandra or in HDFS using CSV or Parquet files. More details on the benchmark are available [here|https://issues.apache.org/jira/browse/CASSANDRA-11542?focusedCommentId=15249213=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15249213] and the code is available [here|https://github.com/stef1927/spark-load-perf]. Here is the comparison with the results of the benchmark that was run on 6th May with 15M rows, as described in [this comment|https://issues.apache.org/jira/browse/CASSANDRA-11542?focusedCommentId=15273884=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15273884]. We can see that the final results are consistent with the proof of concept, which was presented at the Cassandra NGCC conference. !before_after.jpg! * C* old: is TRUNK with no optimizations (at c662d876b95d67a911dfe549d8a0d38ee6fbb904), and the Spark Connector without SPARK-C383 * C* POC: is the [proof-of-concept patch|https://github.com/stef1927/cassandra/commits/9259], and the Spark Connector with an [earlier version|https://github.com/stef1927/spark-cassandra-connector/commits/9259] of SPARK-C383 * C* async: is the CASSANDRA-11521 patch, with results delivered to the client via the new asynchronous paging mechanism * C* sync: is the CASSANDRA-11521 patch, with results delivered to the client via the existing synchronous paging mechanism Here are the results run over several incremental data sets at 15M, 30M, 60M and 120M rows with 256 VNODES: !256_vnodes.jpg! Below are the results run over several incremental data sets at 1 15M, 30M, 60M and 120M rows without VNODES: !no_vnodes.jpg! The raw data is attached [^spark_benchmark_raw_data.zip]. h5. Conclusions * Overall the duration of the 15M row test was improved by 65% (from about 40 to 14 seconds) for schema 1 and by 56% (from 23 to 10 seconds) for schema 3. * The new asynchronous paging mechanism significantly outperforms the existing mechanism with large data sets. For example, for schema 1 and 120M rows, it is approximately 30% faster. In order to achieve this, it is however necessary that the driver reserves a connection per asynchronous paging request, sharing connections degrades performance significantly and makes it no better than the existing mechanism. * CSV still outperforms C* for schema 1 RDD tests. However, for DF tests and schema 3 RDD tests, C* is now on par or faster than CSV. This indicates that the number of cells in CQL rows continues to impact performance. * Parquet is in a league of its own due to its efficient columnar format. It should however be noted that it may be storing the row count in metadata. A more meaningful benchmark could have been obtained had we excluded the row count from the time measurements. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: 256_vnodes.jpg, before_after.jpg, > bulk-read-benchmark.1.html, bulk-read-jfr-profiles.1.tar.gz, > bulk-read-jfr-profiles.2.tar.gz, no_vnodes.jpg, spark_benchmark_raw_data.zip > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15238992#comment-15238992 ] vincent.poncet commented on CASSANDRA-9259: --- In datawarehouse / analytics usecases, you are doing mostly full scans, (hopefully with some predicate pushdown either projection and filtering), but it is reading a big numbers of rows, then doing aggregations. I just want to say that that takes time. So by definition, an analytic query on an OLTP database is always "wrong", in the sense of at the same time of doing the query, data changed, deleted, updated, inserted. So operational analytics is always approximate. So, my point is in operational analytics, CL=1 will be perfectly fine. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15229592#comment-15229592 ] Stefania commented on CASSANDRA-9259: - Thanks. I've created CASSANDRA-11520 and CASSANDRA-11521. I should also have added the link to the POC patch: https://github.com/stef1927/cassandra/commits/9259. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15228185#comment-15228185 ] Sylvain Lebresne commented on CASSANDRA-9259: - bq. In view of these results, I believe an optimized local read path is worth pursuing whilst streaming should also provide benefits I agree, and I suggest creating 2 different sub-tickets for those and starting with that. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15228054#comment-15228054 ] Stefania commented on CASSANDRA-9259: - Here are the results of a proof of concept of an *optimized read path* for local reads at CL 1 and *streaming*. These are the results in *ops/second*: ||Partition Size||Page size||Num. Partitions in the table||Synchronous, no optimization ops/s||Synchronous, with optimization ops/s||Prefetch a page, no optimization ops/s||Prefetch a page, with optimization ops/s||Streaming, measurement 1 ops/s||Streaming, measurement 2 ops/s||Improvement due to local read optimization||Improvement due to streaming||Total improvement||Link to the test|| |100 KBYTES|10|250K|98|148|189|283|784|767|49.74%|174.03%|310.32%|[link|http://cstar.datastax.com/tests/id/6dc252e8-fb0c-11e5-9ad6-0256e416528f]| |10 KBYTES|100|1M|93|138|174|259|659|673|48.85%|157.14%|282.76%|[link|http://cstar.datastax.com/tests/id/d8002fe2-fad3-11e5-a500-0256e416528f]| |1 KBYTE|1000|1M|84|133|114|179|233|239|57.02%|31.84%|107.02%|[link|http://cstar.datastax.com/tests/id/8dd1c5ba-fad9-11e5-82e5-0256e416528f]| |1 KBYTE|1000|2M|60|98|94|153|247|248|62.77%|61.76%|163.30%|[link|http://cstar.datastax.com/tests/id/97be0520-fba6-11e5-bba8-0256e416528f]| |100 BYTES|1|5M|21|33|24|37|41|44|54.17%|14.86%|77.08%|[link|http://cstar.datastax.com/tests/id/95e31c0c-fb12-11e5-9ad6-0256e416528f]| |50 BYTES|2|5M|20|32|20|33|35|37|65.00%|9.09%|80.00%|[link|http://cstar.datastax.com/tests/id/bf1a52a0-fb8c-11e5-838f-0256e416528f]| |10 KBYTES|5000|500K|31|46|31|46|45|46|48.39%|-1.09%|46.77%|[link|http://cstar.datastax.com/tests/id/df9e6e96-fbbc-11e5-bf65-0256e416528f]| |1 KBYTE|5000|2M|35|56|40|64|66|66|60.00%|3.13%|65.00%|[link|http://cstar.datastax.com/tests/id/1d1785fc-fbb6-11e5-bf65-0256e416528f]| |100 BYTES|5000|5M|22|40|31|53|66|66|70.97%|24.53%|112.90%|[link|http://cstar.datastax.com/tests/id/70262d52-fbac-11e5-a876-0256e416528f]| These are the same results but expressed in *rows/second*: ||Partition Size||Page size||Num. Partitions in the table||Synchronous, no optimization rows/s||Synchronous, with optimization rows/s||Prefetch a page, no optimization rows/s||Prefetch a page, with optimization rows/s||Streaming, measurement 1 rows/s||Streaming, measurement 2 rows/s||Improvement due to local read optimization||Improvement due to streaming||Total improvement||Link to the test|| |100 KBYTES|10|250K|963|1453|1849|2761|7702|7522|49.32%|175.70%|311.68%|[link|http://cstar.datastax.com/tests/id/6dc252e8-fb0c-11e5-9ad6-0256e416528f]| |10 KBYTES|100|1M|8830|13159|16572|24591|62649|63975|48.39%|157.46%|282.04%|[link|http://cstar.datastax.com/tests/id/d8002fe2-fad3-11e5-a500-0256e416528f]| |1 KBYTE|1000|1M|52543|83548|71277|112558|145637|150070|57.92%|31.36%|107.44%|[link|http://cstar.datastax.com/tests/id/8dd1c5ba-fad9-11e5-82e5-0256e416528f]| |1 KBYTE|1000|2M|47029|76292|74131|119727|193738|193495|61.51%|61.71%|161.18%|[link|http://cstar.datastax.com/tests/id/97be0520-fba6-11e5-bba8-0256e416528f]| |100 BYTES|1|5M|88130|142800|100590|158780|176699|185292|57.85%|13.99%|79.93%|[link|http://cstar.datastax.com/tests/id/95e31c0c-fb12-11e5-9ad6-0256e416528f]| |50 BYTES|2|5M|94581|153296|97599|157463|169226|175581|61.34%|9.49%|76.64%|[link|http://cstar.datastax.com/tests/id/bf1a52a0-fb8c-11e5-838f-0256e416528f]| |10 KBYTES|5000|500K|14938|22356|15152|22623|22174|22419|49.31%|-1.44%|47.15%|[link|http://cstar.datastax.com/tests/id/df9e6e96-fbbc-11e5-bf65-0256e416528f]| |1 KBYTE|5000|2M|63552|100974|71644|115001|119537|119079|60.52%|3.75%|66.53%|[link|http://cstar.datastax.com/tests/id/1d1785fc-fbb6-11e5-bf65-0256e416528f]| |100 BYTES|5000|5M|70154|126547|98331|168121|208226|207482|70.97%|23.63%|111.38%|[link|http://cstar.datastax.com/tests/id/70262d52-fbac-11e5-a876-0256e416528f]| The columns above refer to the following cassandra-stress operations: *Synchronous page retrieval* The client retrieves each page synchronously, with and without the optimized local read path. *Asynchronous page retrieval (prefetch)* The client retrieves the first page synchronously and then prefetches the next page, before processing the results of the previous page, with and without the optimized local read path. *Streaming* The client requests all pages initially and then waits synchronously for the first page. For the following pages, each operation processes a page that was previously delivered, blocking only if a page is unavailable. There are two equivalent measurements for streaming because the local read path optimization is always available; it would have added considerable extra work to implement streaming without optimized read path, and it would have only provided comparison data which is already available. *Results* The improvement due to local read optimization is calculated by comparing the
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15215303#comment-15215303 ] Stefania commented on CASSANDRA-9259: - Thank you for sharing the code, it will save us a lot of time if we decide to try out different transfer mechanisms. For now, I am working on streaming and the other optimizations as described above; later we may well focus on different transfer mechanisms. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15215297#comment-15215297 ] Stefania commented on CASSANDRA-9259: - Thank you for this observation. Whilst this is a valid point, the focus of this patch is local transfers at CL=1. For CL > 1, the latency introduced by coordinating across C* nodes is probably the dominating factor and this will not be addressed in this patch. Here we are focusing on local transfers at CL=1. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15214212#comment-15214212 ] Gary Dusbabek commented on CASSANDRA-9259: -- FWIW I experimented with domain sockets late last year. Domain sockets were faster, but not much faster than reading for eth0->eth0, which on modern linux distros goes over the loopback (try it). Experimental branches of the Datastax java driver and Cassandra: https://github.com/gdusbabek/cassandra/tree/cassandra-3.0.2-domain-socket and https://github.com/gdusbabek/java-driver/tree/domain_sockets_3.0 > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15209915#comment-15209915 ] Evan Chan commented on CASSANDRA-9259: -- The biggest difference between HDFS and local client reads, say by scanning SSTables directly or going through mmap / domain sockets / other faster local-only mechanism, is consistency. We're not guaranteed that any one of N replicas has the exact same SSTables for any moment in time, are we? Thus, different Spark workers would end up with potentially different data, for the same token range -- essentially CL-1 behavior, which seems not really acceptable to folks used to precision analytics. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15209656#comment-15209656 ] Stefania commented on CASSANDRA-9259: - I would like to make sure we are all on the same page and gather any further suggestions or thoughts. It would also help very much to have an indication of what we are aiming for, in terms of performance. I am not familiar with either EDW or HDFS, so it would be helpful to roughly quantify what's considered good. We can optimize the existing read path by special-casing local reads at CL=1 and by optimizing any other areas highlighted by the analysis above. The underlying classes for reading sstables are the same as those used for compaction, so this is a similar approach to the "streaming compaction" approach mentioned above, provided we can replace CQL paging with streaming. To do this, we can add some options to the {{SELECT}} CQL command. I'm not sure if it's required at all, but we can improve or change the CQL rows output format if needed. The {{SELECT}} options would allow changing both transfer mechanism and output format, if required. In addition to streaming, if local clients are important, we could also consider saving the output to a memory mapped file, or replacing sockets with Unix domain pipes, if tests indicate that these are significantly faster delivery mechanisms for local clients. Some alternatives: * An off-line tool for scanning sstables. This is in principle simple and efficient but it has the disadvantage that it is a new tool, and we already have many tools. Because it requires access to sstables, it would not work very well for clients running on a different host. * A JMX/nodetool command, as above, this is again something new and in order to implement some remote communication, it would require new plumbing, unless we just save data to a file, which again would not be very friendly for remote clients. * A new CQL command, something like {{BULK EXPORT}}. This is very similar to the approach described above. It is more work for the drivers however, and the functionality is similar to {{SELECT}}, except for delivery and _maybe_ output format. I would probably consider this approach, if we are positive that we need a new output format. [~brianmhess], [~rspitzer], [~slebresne] any thoughts? > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15198628#comment-15198628 ] Stefania commented on CASSANDRA-9259: - Thanks! 100k rows/second looks like an upper-bound but my guess would be that it is related to CQL encoding, not so much to how fast we decode data from sstables. For the sake of completeness, we can push more than 100k rows/second by adding more threads to cassandra stress. I'll make sure to add a test case with more columns and one with an even smaller partition to the next set of _cassandra-stress_ tests, so we'll know for sure. It may be advantageous to also have micro benchmarks for decoding sstables and for other parts that become relevant to the final solution that we choose. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15197725#comment-15197725 ] Vassil Lunchev commented on CASSANDRA-9259: --- Very good results! >From the benchmarks it seems like 100k rows/second is something like a limit. >I have seen that limit in tests as well and to me it is more like 100k >cells/second. Do you think Cassandra would be able to push more than 100k rows/second with partition sizes smaller than 100-bytes (I know that is unpractical)? Also do you think adding more columns to the rows will have any effect. Like, do you think the bound is around 100k rows/second or around 100k cells/second? If I have to bet, it would still be around 100k per second even with smaller than 100 bytes partitions, and the bottleneck is the number of cells, not the number of rows. > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Assignee: Stefania >Priority: Critical > Fix For: 3.x > > Attachments: bulk-read-benchmark.1.html, > bulk-read-jfr-profiles.1.tar.gz, bulk-read-jfr-profiles.2.tar.gz > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15197232#comment-15197232 ] Stefania commented on CASSANDRA-9259: - Below are the results of a set of benchmarking measurements that were performed using the _cassandra-stress_ token range queries as delivered by CASSANDRA-10331. A token range query is a query as outlined in the description of this ticket: {{SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND Token(partitionKey) <= Y}}. These are suggested code enhancements that follow from the analysis of Java Flight Recorder profiles taken during these measurements. h5. Optimizations to the current read path * Avoid using heap buffers in {{DataOutputBuffer}}, ideally we should use the buffer pool and the buffer pool should be extended to support sizes larger than 64 kbytes (point 4 of CASSANDRA-9468). * Avoid creating a local response in {{LocalReadRunnable}}, which serializes the partition iterator in the READ stage thread only for it to be deserialized again in the storage proxy thread. I realize this is problematic because we need to guard sstables and memtables via a {{ReadExecutionController}}, as well as guarantee iterator safety, but removing this redundant step should help significantly for large analytics workloads where the client queries data locally. Perhaps this should be done together with a new output format optimized for this workflow, as described further below. Any thoughts [~slebresne]? * Improve access to sstables for local token ranges in order to avoid the binary search on the index summary, and scanning of index entries until the range is found. * Increase parallelism by splitting local token ranges; at the moment there is parallelism in storage proxy {{getRangeSlice()}} but ranges are only split by replica, a local range is not divided into smaller ranges. Splitting a local range into sub-ranges as stored in different sstables should increase performance. h5. Extensive re-alignment of format and transfer mechanism In addition to optimizing the current read path, to avoid the cost of encoding CQL rows, we could replace them with a format that is more analytics friendly. This would be similar to a special compaction task outputting a columnar format optimized for the query as suggested above. This format should allow efficient decoding in the client and its data should be transferred from Cassandra to the client as quickly as possible. If the client runs locally, for example we could use one of the following mechanisms: * shared memory; * Unix domain pipes; * another [fast IPC mechanisms|https://lwn.net/Articles/466304/]. We should probably leverage either [Apache Parquet|https://parquet.apache.org] for the format or [Apache arrow|https://arrow.apache.org] for both format and transfer mechanism. I think the latter is more aligned to what we are trying to do, but unfortunately this project is just starting. As far as I understand, eventually Spark will read Arrow memory and so if we wrote Arrow memory the transfer should become extremely efficient. h5. Benchmark setup Each _cassandra-stress_ operation corresponds to the retrieval of one page using a token range query where all columns are retrieved. Each token range corresponds to a VNODE range and the Cassandra host was configured with 256 ranges. Different partition sizes were tested: 100 bytes, 1 kbyte, 10 kbytes, 100 kbytes and 1 MB. Each partition had a single CQL row; for the 100 kbytes partition size however, two more tests were added, with 100 and 1000 CQL rows respectively. The size of the rows was adjusted so that the partition size stayed at approximately 100 kbytes. The page size for each test was chosen so that the amount of data downloaded in each operation was roughly 1 MB. So we can view the _cassandra-stress_ ops / second as roughly MB / seconds. A single _cassandra-stress_ thread was used. The tests were run on a GCE *n1-standard-8* VM: 4 Intel Xeon @ 2.60GHz physical cores with 2 hyper-threads per core - all cores located on a single socket, 29 GB of Memory and a 1073GB SCSI hard disk. Both _cassandra-stress_ and the Cassandra process were running on the same VM, as I understand this is normally how Spark or other analytics tools are deployed. h5. Benchmark results The following is a summary of the results in ops / second, [^bulk-read-benchmark.1.html] attached also contains diagrams, partitions/second and rows/second. ||100-bytes||1-kbyte||10-kbytes||100-kbytes||100-kbytes-100-rows||100-kbytes-1000-rows||1-MB|| |28|59|101|80|50|10|73| The performance is variable and very much dependent on partition size and number of CQL rows in a partition. Java Flight Recorder (JFR) files for all cases, albeit measured with a smaller sample, have also been attached, see [^bulk-read-jfr-profiles.1.tar.gz] and
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15117397#comment-15117397 ] Vassil Lunchev commented on CASSANDRA-9259: --- "For full data queries it may be advantageous to have C* be able to compact all of the relevant sstables into a format friendlier to analytics workloads." I would even go further and say - "Cassandra needs a new compaction strategy. It has DateTieredCompactionStrategy for time series data. It needs a new one, for example ColumnarCompactionStrategy, that is similar in concept to Parquet and designed for analytics workloads." The results here: https://github.com/velvia/cassandra-gdelt and the ideas here: https://github.com/tuplejump/FiloDB are very compelling. FilloDB is practically doing a new columnar compaction layer on top of C*. And the results are quite promising - "faster than Parquet scan speeds" with storage needs "within 35% of Parquet". > Bulk Reading from Cassandra > --- > > Key: CASSANDRA-9259 > URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 > Project: Cassandra > Issue Type: New Feature > Components: Compaction, CQL, Local Write-Read Paths, Streaming and > Messaging, Testing >Reporter: Brian Hess >Priority: Critical > Fix For: 3.x > > > This ticket is following on from the 2015 NGCC. This ticket is designed to > be a place for discussing and designing an approach to bulk reading. > The goal is to have a bulk reading path for Cassandra. That is, a path > optimized to grab a large portion of the data for a table (potentially all of > it). This is a core element in the Spark integration with Cassandra, and the > speed at which Cassandra can deliver bulk data to Spark is limiting the > performance of Spark-plus-Cassandra operations. This is especially of > importance as Cassandra will (likely) leverage Spark for internal operations > (for example CASSANDRA-8234). > The core CQL to consider is the following: > SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) > X AND > Token(partitionKey) <= Y > Here, we choose X and Y to be contained within one token range (perhaps > considering the primary range of a node without vnodes, for example). This > query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk > operations via Spark (or other processing frameworks - ETL, etc). There are > a few causes (e.g., inefficient paging). > There are a few approaches that could be considered. First, we consider a > new "Streaming Compaction" approach. The key observation here is that a bulk > read from Cassandra is a lot like a major compaction, though instead of > outputting a new SSTable we would output CQL rows to a stream/socket/etc. > This would be similar to a CompactionTask, but would strip out some > unnecessary things in there (e.g., some of the indexing, etc). Predicates and > projections could also be encapsulated in this new "StreamingCompactionTask", > for example. > Another approach would be an alternate storage format. For example, we might > employ Parquet (just as an example) to store the same data as in the primary > Cassandra storage (aka SSTables). This is akin to Global Indexes (an > alternate storage of the same data optimized for a particular query). Then, > Cassandra can choose to leverage this alternate storage for particular CQL > queries (e.g., range scans). > These are just 2 suggestions to get the conversation going. > One thing to note is that it will be useful to have this storage segregated > by token range so that when you extract via these mechanisms you do not get > replications-factor numbers of copies of the data. That will certainly be an > issue for some Spark operations (e.g., counting). Thus, we will want > per-token-range storage (even for single disks), so this will likely leverage > CASSANDRA-6696 (though, we'll want to also consider the single disk case). > It is also worth discussing what the success criteria is here. It is > unlikely to be as fast as EDW or HDFS performance (though, that is still a > good goal), but being within some percentage of that performance should be > set as success. For example, 2x as long as doing bulk operations on HDFS > with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14715707#comment-14715707 ] Evan Volgas commented on CASSANDRA-9259: For full data queries it may be advantageous to have C* be able to compact all of the relevant sstables into a format friendlier to analytics workloads. --- I cannot agree with [~rspitzer] enough on this point. I tried running Cassandra with Presto recently on a handful of r3.2xlarges... even a simple count(*) of a few million rows took 15-20 minutes, most of which was spent deserializing the SSTable for Presto to read it (and that's to say nothing of the insanely high timeout settings that were required for it to work). If you could compact your data to something like Avro/Parquet, you open the doors to a tremendous variety of analytical / ad hoc / exploratory work loads. I would be nothing short of delighted if this were possible in Cassandra. Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: New Feature Components: Core Reporter: Brian Hess Assignee: Ariel Weisberg Priority: Critical Fix For: 3.x This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14641721#comment-14641721 ] Benedict commented on CASSANDRA-9259: - FTR, I very much favour the streaming compaction approach. Compaction should be just about our most optimised code path. If we cannot make it fast enough, nothing will be. If it isn't currently fast enough, we should make it faster. CASSANDRA-8630 and CASSANDRA-9500 are both related. Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: Improvement Components: Core Reporter: Brian Hess Assignee: Ariel Weisberg This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14540321#comment-14540321 ] Russell Alexander Spitzer commented on CASSANDRA-9259: -- For full data queries it may be advantageous to have C* be able to compact all of the relevant sstables into a format friendlier to analytics workloads. For example a {{compactToParquet}} command could be quite useful for doing batch analytics. This would be useful for full table work because: We are going to need to write the data to a new storage format (or spill serialized data to disk anyway) A durable copy of the data in the new format will stop us from worrying about losing spilled blocks during execution (with spark executors dying and their block stores being lost) Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: Improvement Components: Core Reporter: Brian Hess This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14541150#comment-14541150 ] Jeremy Hanna commented on CASSANDRA-9259: - With compaction being single threaded, perhaps we could parallelize to N extract files where N is the number of concurrent compactors configured. That way you have a small number of files but are using multiple cpu cores to make better use of the IO. Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: Improvement Components: Core Reporter: Brian Hess This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14529530#comment-14529530 ] Jonathan Ellis commented on CASSANDRA-9259: --- Also, CASSANDRA-8630 Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: Improvement Components: Core Reporter: Brian Hess This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra
[ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14520416#comment-14520416 ] Robert Stupp commented on CASSANDRA-9259: - Related to CASSANDRA-6167? Bulk Reading from Cassandra --- Key: CASSANDRA-9259 URL: https://issues.apache.org/jira/browse/CASSANDRA-9259 Project: Cassandra Issue Type: Improvement Components: Core Reporter: Brian Hess This ticket is following on from the 2015 NGCC. This ticket is designed to be a place for discussing and designing an approach to bulk reading. The goal is to have a bulk reading path for Cassandra. That is, a path optimized to grab a large portion of the data for a table (potentially all of it). This is a core element in the Spark integration with Cassandra, and the speed at which Cassandra can deliver bulk data to Spark is limiting the performance of Spark-plus-Cassandra operations. This is especially of importance as Cassandra will (likely) leverage Spark for internal operations (for example CASSANDRA-8234). The core CQL to consider is the following: SELECT a, b, c FROM myKs.myTable WHERE Token(partitionKey) X AND Token(partitionKey) = Y Here, we choose X and Y to be contained within one token range (perhaps considering the primary range of a node without vnodes, for example). This query pushes 50K-100K rows/sec, which is not very fast if we are doing bulk operations via Spark (or other processing frameworks - ETL, etc). There are a few causes (e.g., inefficient paging). There are a few approaches that could be considered. First, we consider a new Streaming Compaction approach. The key observation here is that a bulk read from Cassandra is a lot like a major compaction, though instead of outputting a new SSTable we would output CQL rows to a stream/socket/etc. This would be similar to a CompactionTask, but would strip out some unnecessary things in there (e.g., some of the indexing, etc). Predicates and projections could also be encapsulated in this new StreamingCompactionTask, for example. Another approach would be an alternate storage format. For example, we might employ Parquet (just as an example) to store the same data as in the primary Cassandra storage (aka SSTables). This is akin to Global Indexes (an alternate storage of the same data optimized for a particular query). Then, Cassandra can choose to leverage this alternate storage for particular CQL queries (e.g., range scans). These are just 2 suggestions to get the conversation going. One thing to note is that it will be useful to have this storage segregated by token range so that when you extract via these mechanisms you do not get replications-factor numbers of copies of the data. That will certainly be an issue for some Spark operations (e.g., counting). Thus, we will want per-token-range storage (even for single disks), so this will likely leverage CASSANDRA-6696 (though, we'll want to also consider the single disk case). It is also worth discussing what the success criteria is here. It is unlikely to be as fast as EDW or HDFS performance (though, that is still a good goal), but being within some percentage of that performance should be set as success. For example, 2x as long as doing bulk operations on HDFS with similar node count/size/etc. -- This message was sent by Atlassian JIRA (v6.3.4#6332)