Skip to content

GH-3530: Bypass Hadoop codec abstraction to optimize compression performance #3570

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
iemejia wants to merge 1 commit into
base: master
Choose a base branch
from
Open

GH-3530: Bypass Hadoop codec abstraction to optimize compression performance #3570

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions parquet-benchmarks/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,12 @@
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
<dependency>
<groupId>com.aayushatharva.brotli4j</groupId>
<artifactId>brotli4j</artifactId>
<version>${brotli4j.version}</version>
<scope>runtime</scope>
</dependency>
</dependencies>

<build>
Expand All @@ -102,6 +108,9 @@
<version>${jmh.version}</version>
</path>
</annotationProcessorPaths>
<annotationProcessors>
<annotationProcessor>org.openjdk.jmh.generators.BenchmarkProcessor</annotationProcessor>
</annotationProcessors>
</configuration>
</plugin>
<plugin>
Expand All @@ -121,6 +130,12 @@
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.openjdk.jmh.Main</mainClass>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/BenchmarkList</resource>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/CompilerHints</resource>
</transformer>
</transformers>
<artifactSet>
<includes>
Expand Down
12 changes: 11 additions & 1 deletion parquet-benchmarks/run.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ Information on the JMH_OPTIONS can be found by running: run.sh all -help
----------- | ----------
all | Runs all benchmarks in the module (listed here and others).
build | (No benchmark run, shortcut to rebuild the JMH uber jar).
generate | (No benchmark run, shortcut to pre-generate the per-codec read corpus).
clean | (No benchmark run, shortcut to clean up any temporary files).
read | Reading files with different compression, page and block sizes.
write | Writing files.
Expand Down Expand Up @@ -74,9 +75,18 @@ elif [ "$BENCHMARK" == "build" ]; then

elif [ "$BENCHMARK" == "clean" ]; then

# Shortcut utility to clean any state left behind from any previous run.
# Shortcut utility to clean up any state left behind from any previous run.
java -cp ${SCRIPT_PATH}/target/parquet-benchmarks.jar org.apache.parquet.benchmarks.DataGenerator cleanup

elif [ "$BENCHMARK" == "generate" ]; then

# Shortcut utility to pre-generate the per-codec 1M-row data files reused by the read benchmark.
# Build the uberjar first if it isn't there yet.
if [ ! -f ${SCRIPT_PATH}/target/parquet-benchmarks.jar ]; then
${SCRIPT_PATH}/run.sh build
fi
java -cp ${SCRIPT_PATH}/target/parquet-benchmarks.jar org.apache.parquet.benchmarks.DataGenerator generate

else

# Actually run a benchmark in the JMH harness.
Expand Down
13 changes: 0 additions & 13 deletions parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BenchmarkFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,19 +27,6 @@ public class BenchmarkFiles {
public static final String TARGET_DIR = "target/tests/ParquetBenchmarks";
public static final Path targetDir = new Path(TARGET_DIR);

public static final Path file_1M = new Path(TARGET_DIR + "/PARQUET-1M");

// different block and page sizes
public static final Path file_1M_BS256M_PS4M = new Path(TARGET_DIR + "/PARQUET-1M-BS256M_PS4M");
public static final Path file_1M_BS256M_PS8M = new Path(TARGET_DIR + "/PARQUET-1M-BS256M_PS8M");
public static final Path file_1M_BS512M_PS4M = new Path(TARGET_DIR + "/PARQUET-1M-BS512M_PS4M");
public static final Path file_1M_BS512M_PS8M = new Path(TARGET_DIR + "/PARQUET-1M-BS512M_PS8M");

// different compression codecs
// public final Path parquetFile_1M_LZO = new Path("target/tests/ParquetBenchmarks/PARQUET-1M-LZO");
public static final Path file_1M_SNAPPY = new Path(TARGET_DIR + "/PARQUET-1M-SNAPPY");
public static final Path file_1M_GZIP = new Path(TARGET_DIR + "/PARQUET-1M-GZIP");

// Page checksum files
public static final Path file_100K_CHECKSUMS_UNCOMPRESSED =
new Path(TARGET_DIR + "/PARQUET-100K-CHECKSUMS-UNCOMPRESSED");
Expand Down
444 changes: 444 additions & 0 deletions parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java
View file
Open in desktop

Large diffs are not rendered by default.

120 changes: 32 additions & 88 deletions parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/DataGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,30 +19,17 @@
package org.apache.parquet.benchmarks;

import static java.util.UUID.randomUUID;
import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_256M;
import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_512M;
import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_DEFAULT;
import static org.apache.parquet.benchmarks.BenchmarkConstants.DICT_PAGE_SIZE;
import static org.apache.parquet.benchmarks.BenchmarkConstants.FIXED_LEN_BYTEARRAY_SIZE;
import static org.apache.parquet.benchmarks.BenchmarkConstants.ONE_MILLION;
import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_4M;
import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_8M;
import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_DEFAULT;
import static org.apache.parquet.benchmarks.BenchmarkFiles.TARGET_DIR;
import static org.apache.parquet.benchmarks.BenchmarkFiles.configuration;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS4M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS8M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS4M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS8M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_GZIP;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_SNAPPY;
import static org.apache.parquet.benchmarks.BenchmarkFiles.targetDir;
import static org.apache.parquet.benchmarks.BenchmarkUtils.deleteIfExists;
import static org.apache.parquet.benchmarks.BenchmarkUtils.exists;
import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.SNAPPY;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED;
import static org.apache.parquet.schema.MessageTypeParser.parseMessageType;

import java.io.IOException;
Expand All @@ -60,80 +47,13 @@

public class DataGenerator {

public void generateAll() {
try {
generateData(
file_1M,
configuration,
PARQUET_2_0,
BLOCK_SIZE_DEFAULT,
PAGE_SIZE_DEFAULT,
FIXED_LEN_BYTEARRAY_SIZE,
UNCOMPRESSED,
ONE_MILLION);

// generate data for different block and page sizes
generateData(
file_1M_BS256M_PS4M,
configuration,
PARQUET_2_0,
BLOCK_SIZE_256M,
PAGE_SIZE_4M,
FIXED_LEN_BYTEARRAY_SIZE,
UNCOMPRESSED,
ONE_MILLION);
generateData(
file_1M_BS256M_PS8M,
configuration,
PARQUET_2_0,
BLOCK_SIZE_256M,
PAGE_SIZE_8M,
FIXED_LEN_BYTEARRAY_SIZE,
UNCOMPRESSED,
ONE_MILLION);
generateData(
file_1M_BS512M_PS4M,
configuration,
PARQUET_2_0,
BLOCK_SIZE_512M,
PAGE_SIZE_4M,
FIXED_LEN_BYTEARRAY_SIZE,
UNCOMPRESSED,
ONE_MILLION);
generateData(
file_1M_BS512M_PS8M,
configuration,
PARQUET_2_0,
BLOCK_SIZE_512M,
PAGE_SIZE_8M,
FIXED_LEN_BYTEARRAY_SIZE,
UNCOMPRESSED,
ONE_MILLION);

// generate data for different codecs
// generateData(parquetFile_1M_LZO, configuration, PARQUET_2_0, BLOCK_SIZE_DEFAULT, PAGE_SIZE_DEFAULT,
// FIXED_LEN_BYTEARRAY_SIZE, LZO, ONE_MILLION);
generateData(
file_1M_SNAPPY,
configuration,
PARQUET_2_0,
BLOCK_SIZE_DEFAULT,
PAGE_SIZE_DEFAULT,
FIXED_LEN_BYTEARRAY_SIZE,
SNAPPY,
ONE_MILLION);
generateData(
file_1M_GZIP,
configuration,
PARQUET_2_0,
BLOCK_SIZE_DEFAULT,
PAGE_SIZE_DEFAULT,
FIXED_LEN_BYTEARRAY_SIZE,
GZIP,
ONE_MILLION);
} catch (IOException e) {
throw new RuntimeException(e);
}
/**
* Returns the canonical path of the 1M-row benchmark file for the given codec. Shared by
* {@link #generateAll()}, {@link WriteBenchmarks} and {@link ReadBenchmarks} so that a corpus
* built up front can be reused by the read benchmark instead of being regenerated on demand.
*/
static Path benchmarkFile(CompressionCodecName codec) {
return new Path(TARGET_DIR + "/PARQUET-1M-" + codec.name());
}

public void generateData(
Expand Down Expand Up @@ -200,6 +120,30 @@ public void cleanup() {
deleteIfExists(configuration, targetDir);
}

/**
* Pre-generates the 1M-row data file for every {@link CompressionCodecName} at the default block
* and page size. This is optional: {@link ReadBenchmarks} generates whichever file it needs on
* demand (and {@link #generateData} skips files that already exist), but building the whole
* corpus up front lets repeated read runs reuse it instead of regenerating each time.
*/
public void generateAll() {
try {
for (CompressionCodecName codec : CompressionCodecName.values()) {
generateData(
benchmarkFile(codec),
configuration,
PARQUET_2_0,
BLOCK_SIZE_DEFAULT,
PAGE_SIZE_DEFAULT,
FIXED_LEN_BYTEARRAY_SIZE,
codec,
ONE_MILLION);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public static void main(String[] args) {
DataGenerator generator = new DataGenerator();
if (args.length < 1) {
Expand Down
129 changes: 52 additions & 77 deletions parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/ReadBenchmarks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,107 +18,82 @@
*/
package org.apache.parquet.benchmarks;

import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_DEFAULT;
import static org.apache.parquet.benchmarks.BenchmarkConstants.FIXED_LEN_BYTEARRAY_SIZE;
import static org.apache.parquet.benchmarks.BenchmarkConstants.ONE_MILLION;
import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_DEFAULT;
import static org.apache.parquet.benchmarks.BenchmarkFiles.configuration;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS4M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS8M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS4M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS8M;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_GZIP;
import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_SNAPPY;
import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0;

import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.infra.Blackhole;

/**
* End-to-end read benchmark: reads back 1M rows from a real Parquet file for each compression
* codec, measuring the full read path (filesystem I/O + decompression + decoding).
*
* <p>Parameterized by {@code codec} so every codec is covered uniformly. This replaces the
* previous layout that had one hand-written {@code @Benchmark} method per codec/block/page
* combination (and therefore only covered UNCOMPRESSED, SNAPPY, GZIP and LZO). The file for the
* selected codec is generated once per trial in {@link #generateFileForRead()} (skipped if it
* already exists), reading the shared {@link DataGenerator#benchmarkFile corpus} that can be
* pre-built for every codec via {@code DataGenerator generate} (see {@code run.sh generate}). Run
* a subset with e.g. {@code -p codec=ZSTD,LZ4_RAW}.
*/
@State(Scope.Benchmark)
@BenchmarkMode(Mode.SingleShotTime)
public class ReadBenchmarks {

private void read(Path parquetFile, int nRows, Blackhole blackhole) throws IOException {
ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), parquetFile)
.withConf(configuration)
.build();
for (int i = 0; i < nRows; i++) {
Group group = reader.read();
blackhole.consume(group.getBinary("binary_field", 0));
blackhole.consume(group.getInteger("int32_field", 0));
blackhole.consume(group.getLong("int64_field", 0));
blackhole.consume(group.getBoolean("boolean_field", 0));
blackhole.consume(group.getFloat("float_field", 0));
blackhole.consume(group.getDouble("double_field", 0));
blackhole.consume(group.getBinary("flba_field", 0));
blackhole.consume(group.getInt96("int96_field", 0));
}
reader.close();
}

/**
* This needs to be done exactly once. To avoid needlessly regenerating the files for reading, they aren't cleaned
* as part of the benchmark. If the files exist, a message will be printed and they will not be regenerated.
*/
@Setup(Level.Trial)
public void generateFilesForRead() {
new DataGenerator().generateAll();
}

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsDefaultBlockAndPageSizeUncompressed(Blackhole blackhole) throws IOException {
read(file_1M, ONE_MILLION, blackhole);
}
@Param({"UNCOMPRESSED", "SNAPPY", "GZIP", "LZO", "BROTLI", "LZ4", "ZSTD", "LZ4_RAW"})
public String codec;

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsBS256MPS4MUncompressed(Blackhole blackhole) throws IOException {
read(file_1M_BS256M_PS4M, ONE_MILLION, blackhole);
}
private Path file;

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsBS256MPS8MUncompressed(Blackhole blackhole) throws IOException {
read(file_1M_BS256M_PS8M, ONE_MILLION, blackhole);
}

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsBS512MPS4MUncompressed(Blackhole blackhole) throws IOException {
read(file_1M_BS512M_PS4M, ONE_MILLION, blackhole);
}

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsBS512MPS8MUncompressed(Blackhole blackhole) throws IOException {
read(file_1M_BS512M_PS8M, ONE_MILLION, blackhole);
}

// TODO how to handle lzo jar?
// @Benchmark
// public void read1MRowsDefaultBlockAndPageSizeLZO(Blackhole blackhole)
// throws IOException
// {
// read(parquetFile_1M_LZO, ONE_MILLION, blackhole);
// }

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsDefaultBlockAndPageSizeSNAPPY(Blackhole blackhole) throws IOException {
read(file_1M_SNAPPY, ONE_MILLION, blackhole);
@Setup(Level.Trial)
public void generateFileForRead() throws IOException {
CompressionCodecName codecName = CompressionCodecName.valueOf(codec);
file = DataGenerator.benchmarkFile(codecName);
new DataGenerator()
.generateData(
file,
configuration,
PARQUET_2_0,
BLOCK_SIZE_DEFAULT,
PAGE_SIZE_DEFAULT,
FIXED_LEN_BYTEARRAY_SIZE,
codecName,
ONE_MILLION);
}

@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void read1MRowsDefaultBlockAndPageSizeGZIP(Blackhole blackhole) throws IOException {
read(file_1M_GZIP, ONE_MILLION, blackhole);
public void read1MRows(Blackhole blackhole) throws IOException {
try (ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), file)
.withConf(configuration)
.build()) {
for (int i = 0; i < ONE_MILLION; i++) {
Group group = reader.read();
blackhole.consume(group.getBinary("binary_field", 0));
blackhole.consume(group.getInteger("int32_field", 0));
blackhole.consume(group.getLong("int64_field", 0));
blackhole.consume(group.getBoolean("boolean_field", 0));
blackhole.consume(group.getFloat("float_field", 0));
blackhole.consume(group.getDouble("double_field", 0));
blackhole.consume(group.getBinary("flba_field", 0));
blackhole.consume(group.getInt96("int96_field", 0));
}
}
}
}
Loading
Loading