[jira] [Commented] (CASSANDRA-9259) Bulk Reading from Cassandra

["Stefania (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/CASSANDRA-9259?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=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 [^bulk-read-jfr-profiles.2.tar.gz].

h5. Analysis of the JFR profiles

The following broad categories appear as hot-spots in all JFR profiles but with various impacts depending on partition size or number of CQL columns per partition:

* +Allocations outside of TLABs (Thread Local Allocation Buffers)+: for partition sizes bigger than 1 kbyte we have significant memory allocated outside of TLABs caused by heap byte buffers created by {{DataOutputBuffer}} when serializing partitions. Because of the heap buffers, +partition serialization+ becomes a hot-spot in most of the profiles. This is done when creating the local response to the storage proxy thread. 

* +Partition deserialization+: this is expected but the problem is that it is done twice, when reading from sstables and when reading the local response in the storage proxy thread. 

* +CQL encoding+: this become noticeable for small partitions (100 bytes) or large partitions divided into many CQL rows. There are two hot-spots, {{ResultSet.Codec.encode()}} and {{ResultSet.Codec.encodedSize()}}.

* +Reading+: various reading methods appear as hot-spots, especially {{ByteBufferUtil.readUnsignedVInt}}.

* +Uncompression and rebuffering+: it is only noticeable for partitions of 1MB. This is probably because the default compression chunk size and max buffer pool size is 64kbytes.

The following are the top code paths for all profiles:

*100-bytes*

# {{ResultSet.Codec.encode}}, 2.78%, CQL encoding.
# {{BasePartitions.hasNext}}, 2.39%, partition deserialization.
# {{ResultSet.Codec.encodedSize}}, 2.08%, CQL encoding.

*1-kbyte*

# {{ByteBufferUtil.read}}, 2.83%, partition deserialization.
# {{ByteBufferUtil.readUnsignedVInt}}, 2.40%, reading.
# {{HeapByteBuffer}} allocations, 2.37%, allocations outside of TLABs.
# {{ResultSet.Codec.encodedSize}}, 1.94%, CQL encoding.

*10-kbytes*

# {{ByteBufferUtil.read}}, 16.69%, partition deserialization.
# {{HeapByteBuffer}} allocations, 11.27%, allocations outside of TLABs.

*100-kbytes*

# {{ByteBufferUtil.read}}, 32.66%, partition deserialization.
# {{HeapByteBuffer}} allocations, 21.17%, allocations outside of TLABs.


*100-kbytes-100-rows*

# {{ByteBufferUtil.read}}, 4.50%, partition deserialization.
# {{HeapByteBuffer}} allocations, 3.40%, allocations outside of TLABs.
# {{ResultSet.Codec.encodedSize}}, 2.99%, CQL encoding.
# {{RebufferingInputStream.readUnsignedVInt}}, 2.26%, reading.

*100-kbytes-1000-rows*

# {{ResultSet.Codec.encode}}, 5.03%%, CQL encoding.
# {{ResultSet.Codec.encodedSize}}, 3.84%%, CQL encoding.
# {{Cell.Serializer.deserialize}}, 2.06%, partition deserialization.

*1-mbyte*

# {{ByteBufferUtil.read}}, 21.26%, partition deserialization.
# {{LZ4Compressor.uncompress()}}, 8.64%, called by {{BigTableScanner.seekToCurrentRangeStart()}}, uncompression and rebuffering.
# {{HeapByteBuffer}} allocations, 4.86%, allocations outside of TLABs.


> 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)