This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cassandra-easy-stress.git
The following commit(s) were added to refs/heads/main by this push:
new 71b8e74 Compress parquet files further by enabling dictionary
compression and byte stream splitting (#113)
71b8e74 is described below
commit 71b8e7486c1dce025b6249891036d67d39c393a4
Author: dcapwell <[email protected]>
AuthorDate: Fri Oct 30 10:56:26 2026 -0700
Compress parquet files further by enabling dictionary compression and byte
stream splitting (#113)
---
gradle/libs.versions.toml | 2 +-
manual/MANUAL.adoc | 3 +
manual/examples/cassandra-easy-stress-help.txt | 10 ++
.../easystress/collector/AsyncCollector.kt | 7 +-
.../easystress/collector/ParquetCollector.kt | 32 ++++-
.../apache/cassandra/easystress/commands/Run.kt | 11 +-
.../easystress/collector/ParquetCollectorTest.kt | 129 +++++++++++++++------
src/test/resources/log4j2-test.yaml | 7 ++
8 files changed, 157 insertions(+), 44 deletions(-)
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 74c653f..48f20d9 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -26,7 +26,7 @@ prometheus = "0.16.0"
progressbar = "0.7.2"
hdrhistogram = "2.1.12"
agrona = "1.22.0"
-parquet = "1.15.2"
+parquet = "1.17.0"
hadoop = "3.4.3"
junit = "6.1.3"
junit-params = "6.1.3"
diff --git a/manual/MANUAL.adoc b/manual/MANUAL.adoc
index 0d01909..ff44529 100644
--- a/manual/MANUAL.adoc
+++ b/manual/MANUAL.adoc
@@ -327,6 +327,9 @@ This writes operation metrics including operation type,
success/failure status,
If a directory is provided instead of a file, cassandra-easy-stress will
automatically create an appropriately named file in that directory.
+Add `--parquet-byte-stream-split` to use `BYTE_STREAM_SPLIT` encoding for the
`*_duration_ns` columns, which makes the file smaller.
+It is off by default because the reader must support `BYTE_STREAM_SPLIT` on
`INT64` columns (e.g. DuckDB 2.0+ or pyarrow 16+; Spark's vectorized reader
does not).
+
==== Parquet Schema Reference
Each row in the Parquet file represents a single operation with the following
columns:
diff --git a/manual/examples/cassandra-easy-stress-help.txt
b/manual/examples/cassandra-easy-stress-help.txt
index 0a01039..d666804 100644
--- a/manual/examples/cassandra-easy-stress-help.txt
+++ b/manual/examples/cassandra-easy-stress-help.txt
@@ -99,6 +99,12 @@ Usage: cassandra-easy-stress [options] [command] [command
options]
specified path. If the file is a directory, the file will be
named rawlog.parquet within that directory
Default: <empty string>
+ --parquet-byte-stream-split
+ Use BYTE_STREAM_SPLIT encoding for the duration columns of the
+ --parquet file, making it smaller. The reader must support
+ BYTE_STREAM_SPLIT on INT64 columns, e.g. DuckDB >= 2.0 or pyarrow
+ >= 16; Spark's vectorized reader does not support it.
+ Default: false
--partitiongenerator, --pg
Method of generating partition keys. Supports random, normal
(gaussian), and sequence.
@@ -120,6 +126,10 @@ Usage: cassandra-easy-stress [options] [command] [command
options]
default with CASSANDRA_EASY_STRESS_PROM_PORT, or set to 0 to
disable.
Default: 9500
+ --protocol-compression
+ Protocol compression algorithm ('lz4' or 'none'). Default is
+ 'lz4'.
+ Default: lz4
--queue
Queue Depth. 2x the rate by default.
Default: 10000
diff --git
a/src/main/kotlin/org/apache/cassandra/easystress/collector/AsyncCollector.kt
b/src/main/kotlin/org/apache/cassandra/easystress/collector/AsyncCollector.kt
index 80832c1..2a07dc2 100644
---
a/src/main/kotlin/org/apache/cassandra/easystress/collector/AsyncCollector.kt
+++
b/src/main/kotlin/org/apache/cassandra/easystress/collector/AsyncCollector.kt
@@ -15,10 +15,13 @@ import java.util.concurrent.atomic.AtomicInteger
* Base type for collectors that have "expensive" work that needs to happen
off-thread. Every call to collect will
* generate an object, push to a queue, and processed in another worker thread.
*
- * Implementations must define the writer interface which will be called to do
the "real" work for the collector.
+ * Implementations must provide a writer factory; the writer will be called to
do the "real" work for the collector.
+ * A factory (rather than an overridable method) is used since the writer is
created during construction, before any
+ * subclass properties would be initialized.
*/
abstract class AsyncCollector(
fileOrDirectory: File,
+ createWriter: (File) -> Writer,
) : Collector {
data class Event(
val op: Operation,
@@ -48,8 +51,6 @@ abstract class AsyncCollector(
val dropped = AtomicInteger()
val counter = AtomicInteger()
- abstract fun createWriter(fileOrDirectory: File): Writer
-
override fun collect(
ctx: StressContext,
op: Operation,
diff --git
a/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
b/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
index 959681b..89b97e6 100644
---
a/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
+++
b/src/main/kotlin/org/apache/cassandra/easystress/collector/ParquetCollector.kt
@@ -20,12 +20,18 @@ import java.util.concurrent.TimeUnit
*/
class ParquetCollector(
fileOrDirectory: File,
-) : AsyncCollector(fileOrDirectory) {
- override fun createWriter(fileOrDirectory: File): Writer =
- ParquetTableWriter(if (fileOrDirectory.isDirectory)
File(fileOrDirectory, "rawlog.parquet") else fileOrDirectory)
-
+ byteStreamSplit: Boolean = false,
+) : AsyncCollector(
+ fileOrDirectory,
+ { ParquetTableWriter(if (it.isDirectory) File(it, "rawlog.parquet")
else it, byteStreamSplit) },
+ ) {
+ /**
+ * @param byteStreamSplit use BYTE_STREAM_SPLIT for the `*_duration_ns`
columns. Smaller files, but readers must
+ * support BYTE_STREAM_SPLIT on INT64 (e.g. DuckDB >= 2.0, pyarrow >= 16;
not Spark's vectorized reader).
+ */
class ParquetTableWriter(
file: File,
+ byteStreamSplit: Boolean = false,
) : Writer {
private val writer: ParquetWriter<Group>
private val groupFactory: SimpleGroupFactory
@@ -51,13 +57,23 @@ class ParquetCollector(
// Defaults are UNCOMPRESSED + PARQUET_1_0 (PLAIN int64), which
for long runs produces multi-GB files.
// ZSTD + PARQUET_2_0 (DELTA_BINARY_PACKED for the time/duration
columns) is ~3x smaller with the same
// schema and values.
- writer =
+ val builder =
ExampleParquetWriter
.builder(path)
.withType(schema)
.withCompressionCodec(CompressionCodecName.ZSTD)
.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0)
- .build()
+ .withDictionaryEncoding(true)
+ // Large pages let the dictionary hold more distinct stack
traces before falling back to plain.
+ .withPageSize(PAGE_SIZE_BYTES)
+ .withDictionaryPageSize(PAGE_SIZE_BYTES)
+ // start times are near-monotonic and durations near-unique, so
skip the dictionary and go straight to
+ // DELTA_BINARY_PACKED (or BYTE_STREAM_SPLIT for durations, when
enabled).
+ INT64_COLUMNS.forEach { builder.withDictionaryEncoding(it, false) }
+ if (byteStreamSplit) {
+ DURATION_COLUMNS.forEach {
builder.withByteStreamSplitEncoding(it, true) }
+ }
+ writer = builder.build()
}
override fun write(event: Event) {
@@ -88,6 +104,10 @@ class ParquetCollector(
}
companion object {
+ private const val PAGE_SIZE_BYTES = 8 * 1024 * 1024
+ private val DURATION_COLUMNS = listOf("service_duration_ns",
"request_duration_ns")
+ private val INT64_COLUMNS = listOf("service_start_time_ms",
"request_start_time_ms") + DURATION_COLUMNS
+
private fun op(event: Event) =
event.op.javaClass.simpleName
.replace("Statement", "")
diff --git a/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
b/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
index bd37eae..1ab0eb7 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
@@ -293,6 +293,15 @@ class Run(
)
var parquetFile = ""
+ @Parameter(
+ names = ["--parquet-byte-stream-split"],
+ description =
+ "Use BYTE_STREAM_SPLIT encoding for the duration columns of the
--parquet file, making it smaller. " +
+ "The reader must support BYTE_STREAM_SPLIT on INT64 columns,
e.g. DuckDB >= 2.0 or pyarrow >= 16; " +
+ "Spark's vectorized reader does not support it.",
+ )
+ var parquetByteStreamSplit = false
+
@Parameter(names = ["--paging"], description = "Override the driver's
default page size.")
var paging: Int? = null
@@ -585,7 +594,7 @@ class Run(
collectors.add(HdrCollector(hdrHistogramPrefix))
}
if (parquetFile != "") {
- collectors.add(ParquetCollector(File(parquetFile)))
+ collectors.add(ParquetCollector(File(parquetFile),
parquetByteStreamSplit))
}
if (collectors.size == 1) {
return collectors[0]
diff --git
a/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
b/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
index 2b97ac7..eaebb2f 100644
---
a/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
+++
b/src/test/kotlin/org/apache/cassandra/easystress/collector/ParquetCollectorTest.kt
@@ -18,11 +18,15 @@ import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.parquet.hadoop.ParquetReader
import org.apache.parquet.hadoop.example.GroupReadSupport
import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import org.apache.parquet.hadoop.metadata.ParquetMetadata
import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.apache.parquet.schema.MessageTypeParser
import org.assertj.core.api.Assertions
import org.assertj.core.api.Condition
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.io.TempDir
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import java.io.File
import java.nio.file.Path
@@ -63,58 +67,117 @@ internal class ParquetCollectorTest {
)
}
- @Test
- fun usesZstdAndV2Encodings() {
- val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ private fun writeEvents(
+ file: File,
+ byteStreamSplit: Boolean,
+ ): Operation {
val op = Operation.SelectStatement(mockk<BoundStatement>())
- val writer = ParquetCollector.ParquetTableWriter(file)
- val rows = 100
- for (i in 0 until rows) {
- writer.write(
- AsyncCollector.Event(
- op,
- Either.Left(mockk<AsyncResultSet>()),
- op.createdAtNanos + i,
- op.createdAtNanos + 1_000 + i * 2L,
- ),
- )
+ val writer = ParquetCollector.ParquetTableWriter(file, byteStreamSplit)
+ for (i in 0 until ROWS) {
+ val result: Either<AsyncResultSet, Throwable> =
+ if (i % 10 == 9) Either.Right(IllegalStateException("boom
$i")) else Either.Left(mockk<AsyncResultSet>())
+ writer.write(AsyncCollector.Event(op, result, op.createdAtNanos +
i, op.createdAtNanos + 1_000 + i * 2L))
}
writer.close()
+ return op
+ }
- val path =
- org.apache.hadoop.fs
- .Path(file.toURI())
- val conf = Configuration()
- ParquetFileReader.open(HadoopInputFile.fromPath(path, conf)).use {
reader ->
- val columns = reader.footer.blocks.flatMap { it.columns }
- Assertions.assertThat(columns).isNotEmpty
- Assertions.assertThat(columns.map { it.codec
}).containsOnly(CompressionCodecName.ZSTD)
- val int64Encodings =
- columns
- .filter { it.path.toDotString().endsWith("_ms") ||
it.path.toDotString().endsWith("_ns") }
- .flatMap { it.encodings }
-
Assertions.assertThat(int64Encodings).contains(Encoding.DELTA_BINARY_PACKED)
- }
+ private fun hadoopPath(file: File) =
+ org.apache.hadoop.fs
+ .Path(file.toURI())
+
+ private fun footer(file: File): ParquetMetadata =
+ ParquetFileReader.open(HadoopInputFile.fromPath(hadoopPath(file),
Configuration())).use { it.footer }
+
+ private fun encodingsByColumn(footer: ParquetMetadata): Map<String,
Set<Encoding>> =
+ footer.blocks
+ .flatMap { it.columns }
+ .groupBy { it.path.toDotString() }
+ .mapValues { (_, chunks) -> chunks.flatMap { it.encodings
}.toSet() }
+
+ @ParameterizedTest
+ @ValueSource(booleans = [false, true])
+ fun roundTrip(byteStreamSplit: Boolean) {
+ val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ val op = writeEvents(file, byteStreamSplit)
- // semantics must not change: same columns, same values
val read = mutableListOf<Group>()
- ParquetReader.builder(GroupReadSupport(),
path).withConf(conf).build().use { reader ->
+ ParquetReader.builder(GroupReadSupport(),
hadoopPath(file)).withConf(Configuration()).build().use { reader ->
while (true) {
read.add(reader.read() ?: break)
}
}
- Assertions.assertThat(read).hasSize(rows)
+ Assertions.assertThat(read).hasSize(ROWS)
read.forEachIndexed { i, g ->
+ val failed = i % 10 == 9
Assertions.assertThat(g.getString("operation",
0)).isEqualTo("Select")
- Assertions.assertThat(g.getBoolean("success", 0)).isTrue
- Assertions.assertThat(g.getString("failure_reason", 0)).isEmpty()
+ Assertions.assertThat(g.getBoolean("success",
0)).isEqualTo(!failed)
+ if (failed) {
+ Assertions.assertThat(g.getString("failure_reason",
0)).isEqualTo("IllegalStateException")
+ Assertions.assertThat(g.getString("failure_stacktrace",
0)).contains("IllegalStateException: boom $i")
+ } else {
+ Assertions.assertThat(g.getString("failure_reason",
0)).isEmpty()
+ Assertions.assertThat(g.getString("failure_stacktrace",
0)).isEmpty()
+ }
Assertions.assertThat(g.getLong("request_start_time_ms",
0)).isEqualTo(op.createdAtMillis)
Assertions.assertThat(g.getLong("request_duration_ns",
0)).isEqualTo(1_000 + i * 2L)
Assertions.assertThat(g.getLong("service_duration_ns",
0)).isEqualTo(1_000 + i * 2L - i)
}
}
+ @Test
+ fun defaultEncodings() {
+ val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ writeEvents(file, byteStreamSplit = false)
+
+ val footer = footer(file)
+ Assertions.assertThat(footer.blocks.flatMap { it.columns }.map {
it.codec }).containsOnly(CompressionCodecName.ZSTD)
+ val encodings = encodingsByColumn(footer)
+ for (col in INT64_COLUMNS) {
+
Assertions.assertThat(encodings[col]).describedAs(col).contains(Encoding.DELTA_BINARY_PACKED)
+ }
+
Assertions.assertThat(encodings.values.flatten()).doesNotContain(Encoding.BYTE_STREAM_SPLIT)
+ }
+
+ @Test
+ fun byteStreamSplitEncodings() {
+ val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ writeEvents(file, byteStreamSplit = true)
+
+ val encodings = encodingsByColumn(footer(file))
+ for (col in listOf("service_duration_ns", "request_duration_ns")) {
+
Assertions.assertThat(encodings[col]).describedAs(col).contains(Encoding.BYTE_STREAM_SPLIT)
+ }
+ for (col in listOf("service_start_time_ms", "request_start_time_ms")) {
+
Assertions.assertThat(encodings[col]).describedAs(col).contains(Encoding.DELTA_BINARY_PACKED)
+ }
+ }
+
+ /** The output must stay readable by consumers of the released schema. */
+ @Test
+ fun schemaMatchesReleasedSchema() {
+ val released =
+ MessageTypeParser.parseMessageType(
+ "message example {\n" +
+ " required binary operation (UTF8);\n" +
+ " required boolean success;\n" +
+ " required binary failure_reason (UTF8);\n" +
+ " required binary failure_stacktrace (UTF8);\n" +
+ " required int64 service_start_time_ms;\n" +
+ " required int64 service_duration_ns;\n" +
+ " required int64 request_start_time_ms;\n" +
+ " required int64 request_duration_ns;\n" +
+ "}",
+ )
+ val file = tempDir!!.resolve("rawlog.parquet").toFile()
+ writeEvents(file, byteStreamSplit = false)
+
Assertions.assertThat(footer(file).fileMetaData.schema).isEqualTo(released)
+ }
+
companion object {
+ private const val ROWS = 100
+ private val INT64_COLUMNS = listOf("service_start_time_ms",
"service_duration_ns", "request_start_time_ms", "request_duration_ns")
+
val ctx =
Context(
mockk<CqlSession>(),
diff --git a/src/test/resources/log4j2-test.yaml
b/src/test/resources/log4j2-test.yaml
index 468ee43..6bdf7f0 100644
--- a/src/test/resources/log4j2-test.yaml
+++ b/src/test/resources/log4j2-test.yaml
@@ -43,6 +43,13 @@ Configuration:
logger:
+ # parquet/hadoop at debug log every value written, which can produce GBs
of logs for large files
+ -
+ name: org.apache.parquet
+ level: warn
+ -
+ name: org.apache.hadoop
+ level: warn
-
name: org.apache.cassandra.easystress
level: debug
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
