You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@orc.apache.org by pg...@apache.org on 2021/04/08 09:31:13 UTC
[orc] branch master updated: ORC-777: Make the vectorized row batch
size configurable in MR record readers and writers (#674)
This is an automated email from the ASF dual-hosted git repository.
pgaref pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/orc.git
The following commit(s) were added to refs/heads/master by this push:
new fc4cf4c ORC-777: Make the vectorized row batch size configurable in MR record readers and writers (#674)
fc4cf4c is described below
commit fc4cf4c57fa30c3eb684cd63d7658d966821852c
Author: Sudarshan Vasudevan <sv...@users.noreply.github.com>
AuthorDate: Thu Apr 8 02:31:05 2021 -0700
ORC-777: Make the vectorized row batch size configurable in MR record readers and writers (#674)
### What changes were proposed in this pull request?
The changes in this PR create overloaded constructors in MapreduceRecordReader/Writer classes that accept the rowbatch size as an argument. In order to use the overloaded constructor, the caller is expected to override the getRecordWriter() method (or getRecordReader()), as exemplified in the TestMapreduceOrcOutputFormat class.
### Why are the changes needed?
The change is needed to provide greater configurability of the ORC MR readers/writers.
Current behavior of OrcInput(Output)Format is to create an ORC reader/writer with the default row batch size of 1024. This may need to be overridden in certain scenarios to prevent OOMs during reads/writes.
### How was this patch tested?
Added unit tests in TestOrcOutputFormat and TestMapreduceOrcOutputFormat, where we extend the OrcOutputFormat class to override the getRecordWriter() method. The change to override the row batch size config was also tested by running a MR job that used the snapshot version of the orc-mapreduce library.
---
.../apache/orc/mapred/OrcMapredRecordReader.java | 8 +++-
.../apache/orc/mapred/OrcMapredRecordWriter.java | 7 ++-
.../orc/mapreduce/OrcMapreduceRecordReader.java | 8 +++-
.../orc/mapreduce/OrcMapreduceRecordWriter.java | 7 ++-
.../org/apache/orc/mapred/TestOrcOutputFormat.java | 50 ++++++++++++++++++++++
.../mapreduce/TestMapreduceOrcOutputFormat.java | 50 ++++++++++++++++++++++
6 files changed, 126 insertions(+), 4 deletions(-)
diff --git a/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordReader.java b/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordReader.java
index 0a58774..f45f437 100644
--- a/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordReader.java
+++ b/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordReader.java
@@ -70,13 +70,19 @@ public class OrcMapredRecordReader<V extends WritableComparable>
protected OrcMapredRecordReader(Reader fileReader,
Reader.Options options) throws IOException {
+ this(fileReader, options, VectorizedRowBatch.DEFAULT_SIZE);
+ }
+
+ protected OrcMapredRecordReader(Reader fileReader,
+ Reader.Options options,
+ int rowBatchSize) throws IOException {
this.batchReader = fileReader.rows(options);
if (options.getSchema() == null) {
schema = fileReader.getSchema();
} else {
schema = options.getSchema();
}
- this.batch = schema.createRowBatch();
+ this.batch = schema.createRowBatch(rowBatchSize);
rowInBatch = 0;
}
diff --git a/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordWriter.java b/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordWriter.java
index 1a6fda6..da54f0e 100644
--- a/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordWriter.java
+++ b/java/mapreduce/src/java/org/apache/orc/mapred/OrcMapredRecordWriter.java
@@ -63,9 +63,14 @@ public class OrcMapredRecordWriter<V extends Writable>
private final boolean isTopStruct;
public OrcMapredRecordWriter(Writer writer) {
+ this(writer, VectorizedRowBatch.DEFAULT_SIZE);
+ }
+
+ public OrcMapredRecordWriter(Writer writer,
+ int rowBatchSize) {
this.writer = writer;
schema = writer.getSchema();
- this.batch = schema.createRowBatch();
+ this.batch = schema.createRowBatch(rowBatchSize);
isTopStruct = schema.getCategory() == TypeDescription.Category.STRUCT;
}
diff --git a/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordReader.java b/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordReader.java
index 21a5e01..8447d7e 100644
--- a/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordReader.java
+++ b/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordReader.java
@@ -56,13 +56,19 @@ public class OrcMapreduceRecordReader<V extends WritableComparable>
public OrcMapreduceRecordReader(Reader fileReader,
Reader.Options options) throws IOException {
+ this(fileReader, options, VectorizedRowBatch.DEFAULT_SIZE);
+ }
+
+ public OrcMapreduceRecordReader(Reader fileReader,
+ Reader.Options options,
+ int rowBatchSize) throws IOException {
this.batchReader = fileReader.rows(options);
if (options.getSchema() == null) {
schema = fileReader.getSchema();
} else {
schema = options.getSchema();
}
- this.batch = schema.createRowBatch();
+ this.batch = schema.createRowBatch(rowBatchSize);
rowInBatch = 0;
this.row = (V) OrcStruct.createValue(schema);
}
diff --git a/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordWriter.java b/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordWriter.java
index 9379584..287aa5b 100644
--- a/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordWriter.java
+++ b/java/mapreduce/src/java/org/apache/orc/mapreduce/OrcMapreduceRecordWriter.java
@@ -41,9 +41,14 @@ public class OrcMapreduceRecordWriter<V extends Writable>
private final boolean isTopStruct;
public OrcMapreduceRecordWriter(Writer writer) {
+ this(writer, VectorizedRowBatch.DEFAULT_SIZE);
+ }
+
+ public OrcMapreduceRecordWriter(Writer writer,
+ int rowBatchSize) {
this.writer = writer;
schema = writer.getSchema();
- this.batch = schema.createRowBatch();
+ this.batch = schema.createRowBatch(rowBatchSize);
isTopStruct = schema.getCategory() == TypeDescription.Category.STRUCT;
}
diff --git a/java/mapreduce/src/test/org/apache/orc/mapred/TestOrcOutputFormat.java b/java/mapreduce/src/test/org/apache/orc/mapred/TestOrcOutputFormat.java
index 4004681..f570c37 100644
--- a/java/mapreduce/src/test/org/apache/orc/mapred/TestOrcOutputFormat.java
+++ b/java/mapreduce/src/test/org/apache/orc/mapred/TestOrcOutputFormat.java
@@ -42,11 +42,13 @@ import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.util.Progressable;
import org.apache.orc.CompressionKind;
import org.apache.orc.OrcConf;
import org.apache.orc.OrcFile;
import org.apache.orc.Reader;
import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
import org.junit.Test;
import java.io.File;
@@ -295,4 +297,52 @@ public class TestOrcOutputFormat {
assertEquals(3000, file.getNumberOfRows());
assertEquals(TYPE_STRING, file.getSchema().toString());
}
+
+ /**
+ * Make sure that the ORC writer is initialized with a configured row batch size
+ * @throws Exception
+ */
+ @Test
+ public void testOrcOutputFormatWithRowBatchSize() throws Exception {
+ conf.set("mapreduce.output.fileoutputformat.outputdir", workDir.toString());
+ conf.set("mapreduce.task.attempt.id", "attempt_jt0_0_m_0_0");
+ conf.setInt(OrcOutputFormatWithRowBatchSize.ROW_BATCH_SIZE, 128);
+ String TYPE_STRING = "struct<i:int,s:string>";
+ OrcConf.MAPRED_OUTPUT_SCHEMA.setString(conf, TYPE_STRING);
+ conf.setOutputCommitter(NullOutputCommitter.class);
+ TypeDescription schema = TypeDescription.fromString(TYPE_STRING);
+ OrcKey key = new OrcKey(new OrcStruct(schema));
+ RecordWriter<NullWritable, Writable> writer =
+ new OrcOutputFormatWithRowBatchSize<>().getRecordWriter(fs, conf, "key.orc",
+ Reporter.NULL);
+ NullWritable nada = NullWritable.get();
+ for(int r=0; r < 2000; ++r) {
+ ((OrcStruct) key.key).setAllFields(new IntWritable(r),
+ new Text(Integer.toString(r)));
+ writer.write(nada, key);
+ }
+ writer.close(Reporter.NULL);
+ Path path = new Path(workDir, "key.orc");
+ Reader file = OrcFile.createReader(path, OrcFile.readerOptions(conf));
+ assertEquals(2000, file.getNumberOfRows());
+ assertEquals(TYPE_STRING, file.getSchema().toString());
+ }
+
+ private static class OrcOutputFormatWithRowBatchSize<V extends Writable> extends OrcOutputFormat {
+ public static final String ROW_BATCH_SIZE = "orc.row.batch.size";
+
+ @Override
+ public RecordWriter<NullWritable, V> getRecordWriter(FileSystem fileSystem,
+ JobConf conf,
+ String name,
+ Progressable progressable
+ ) throws IOException {
+ Path path = getTaskOutputPath(conf, name);
+ Writer writer = OrcFile.createWriter(path,
+ buildOptions(conf).fileSystem(fileSystem));
+ //Ensure that orc.row.batch.size confing is set in the JobConf
+ int rowBatchSize = Integer.parseInt(conf.get(ROW_BATCH_SIZE));
+ return new OrcMapredRecordWriter<>(writer, rowBatchSize);
+ }
+ }
}
diff --git a/java/mapreduce/src/test/org/apache/orc/mapreduce/TestMapreduceOrcOutputFormat.java b/java/mapreduce/src/test/org/apache/orc/mapreduce/TestMapreduceOrcOutputFormat.java
index 21c96d8..547aec2 100644
--- a/java/mapreduce/src/test/org/apache/orc/mapreduce/TestMapreduceOrcOutputFormat.java
+++ b/java/mapreduce/src/test/org/apache/orc/mapreduce/TestMapreduceOrcOutputFormat.java
@@ -18,6 +18,7 @@
package org.apache.orc.mapreduce;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
@@ -39,6 +40,7 @@ import org.apache.orc.OrcConf;
import org.apache.orc.OrcFile;
import org.apache.orc.Reader;
import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
import org.apache.orc.mapred.OrcKey;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapred.OrcValue;
@@ -282,4 +284,52 @@ public class TestMapreduceOrcOutputFormat {
assertEquals(3000, file.getNumberOfRows());
assertEquals(TYPE_STRING, file.getSchema().toString());
}
+
+ /**
+ * Make sure that the ORC writer is initialized with a configured row batch size
+ * @throws Exception
+ */
+ @Test
+ public void testOrcOutputFormatWithRowBatchSize() throws Exception {
+ conf.set("mapreduce.output.fileoutputformat.outputdir", workDir.toString());
+ conf.setInt(OrcOutputFormatWithRowBatchSize.ROW_BATCH_SIZE, 128);
+ String TYPE_STRING = "struct<i:int,s:string>";
+ OrcConf.MAPRED_OUTPUT_SCHEMA.setString(conf, TYPE_STRING);
+ conf.setBoolean(OrcOutputFormat.SKIP_TEMP_DIRECTORY, true);
+ TaskAttemptID id = new TaskAttemptID("jt", 0, TaskType.MAP, 0, 1);
+ TaskAttemptContext attemptContext = new TaskAttemptContextImpl(conf, id);
+ TypeDescription schema = TypeDescription.fromString(TYPE_STRING);
+ OrcKey key = new OrcKey(new OrcStruct(schema));
+ RecordWriter<NullWritable, Writable> writer =
+ new OrcOutputFormatWithRowBatchSize<>().getRecordWriter(attemptContext);
+ NullWritable nada = NullWritable.get();
+ for(int r=0; r < 2000; ++r) {
+ ((OrcStruct) key.key).setAllFields(new IntWritable(r),
+ new Text(Integer.toString(r)));
+ writer.write(nada, key);
+ }
+ writer.close(attemptContext);
+ Path path = new Path(workDir, "part-m-00000.orc");
+ Reader file = OrcFile.createReader(path, OrcFile.readerOptions(conf));
+ assertEquals(2000, file.getNumberOfRows());
+ assertEquals(TYPE_STRING, file.getSchema().toString());
+ }
+
+ private static class OrcOutputFormatWithRowBatchSize<V extends Writable> extends OrcOutputFormat {
+ private static final String EXTENSION = ".orc";
+ public static final String ROW_BATCH_SIZE = "orc.row.batch.size";
+
+ @Override
+ public RecordWriter<NullWritable, V>
+ getRecordWriter(TaskAttemptContext taskAttemptContext
+ ) throws IOException {
+ Configuration conf = taskAttemptContext.getConfiguration();
+ Path filename = getDefaultWorkFile(taskAttemptContext, EXTENSION);
+ Writer writer = OrcFile.createWriter(filename,
+ org.apache.orc.mapred.OrcOutputFormat.buildOptions(conf));
+ //Ensure that orc.row.batch.size confing is set in the JobConf
+ int rowBatchSize = Integer.parseInt(conf.get(ROW_BATCH_SIZE));
+ return new OrcMapreduceRecordWriter<>(writer, rowBatchSize);
+ }
+ }
}