You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by le...@apache.org on 2020/04/16 23:06:06 UTC
[incubator-gobblin] branch master updated: [GOBBLIN-1117] Enable
record count verification for ORC format
This is an automated email from the ASF dual-hosted git repository.
lesun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 32f866b [GOBBLIN-1117] Enable record count verification for ORC format
32f866b is described below
commit 32f866b08e524148b1c13192a35bc19c3d8b3ee2
Author: Lei Sun <le...@linkedin.com>
AuthorDate: Thu Apr 16 16:05:57 2020 -0700
[GOBBLIN-1117] Enable record count verification for ORC format
Closes #2957 from autumnust/orc-recompact-fix
---
.../compaction/source/CompactionSource.java | 2 +-
.../verify/CompactionThresholdVerifier.java | 31 ++++++++-----
.../compaction/verify/InputRecordCountHelper.java | 12 +++--
.../mapreduce/AvroCompactionTaskTest.java | 2 +-
.../mapreduce/OrcCompactionTaskTest.java | 54 ++++++++++++++++------
5 files changed, 70 insertions(+), 31 deletions(-)
diff --git a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/source/CompactionSource.java b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/source/CompactionSource.java
index 6b7f551..35fe53d 100644
--- a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/source/CompactionSource.java
+++ b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/source/CompactionSource.java
@@ -156,7 +156,7 @@ public class CompactionSource implements WorkUnitStreamSource<String, String> {
Stopwatch stopwatch = Stopwatch.createStarted();
int threads = this.state.getPropAsInt(CompactionVerifier.COMPACTION_VERIFICATION_THREADS, 5);
long timeOutInMinute = this.state.getPropAsLong(CompactionVerifier.COMPACTION_VERIFICATION_TIMEOUT_MINUTES, 30);
- long iterationCountLimit = this.state.getPropAsLong(CompactionVerifier.COMPACTION_VERIFICATION_ITERATION_COUNT_LIMIT, Integer.MAX_VALUE);
+ long iterationCountLimit = this.state.getPropAsLong(CompactionVerifier.COMPACTION_VERIFICATION_ITERATION_COUNT_LIMIT, 100);
long iteration = 0;
Map<String, String> failedReasonMap = null;
while (datasets.size() > 0 && iteration++ < iterationCountLimit) {
diff --git a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/CompactionThresholdVerifier.java b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/CompactionThresholdVerifier.java
index 0eed686..5154afd 100644
--- a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/CompactionThresholdVerifier.java
+++ b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/CompactionThresholdVerifier.java
@@ -32,11 +32,11 @@ import org.apache.gobblin.compaction.conditions.RecompactionConditionBasedOnRati
import org.apache.gobblin.compaction.mapreduce.MRCompactor;
import org.apache.gobblin.compaction.parser.CompactionPathParser;
import org.apache.gobblin.configuration.State;
-import org.apache.gobblin.data.management.dataset.SimpleFileSystemDataset;
import org.apache.gobblin.dataset.FileSystemDataset;
+
/**
- * Compare the source and destination avro records. Determine if a compaction is needed.
+ * Compare the source and destination file records' count. Determine if a compaction is needed.
*/
@Slf4j
public class CompactionThresholdVerifier implements CompactionVerifier<FileSystemDataset> {
@@ -62,34 +62,41 @@ public class CompactionThresholdVerifier implements CompactionVerifier<FileSyste
*
* @return true iff the difference exceeds the threshold or this is the first time compaction
*/
- public Result verify (FileSystemDataset dataset) {
+ public Result verify(FileSystemDataset dataset) {
Map<String, Double> thresholdMap = RecompactionConditionBasedOnRatio.
- getDatasetRegexAndRecompactThreshold (state.getProp(MRCompactor.COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET,
- StringUtils.EMPTY));
+ getDatasetRegexAndRecompactThreshold(
+ state.getProp(MRCompactor.COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET, StringUtils.EMPTY));
CompactionPathParser.CompactionParserResult result = new CompactionPathParser(state).parse(dataset);
- double threshold = RecompactionConditionBasedOnRatio.getRatioThresholdByDatasetName(result.getDatasetName(), thresholdMap);
- log.debug ("Threshold is {} for dataset {}", threshold, result.getDatasetName());
+ double threshold =
+ RecompactionConditionBasedOnRatio.getRatioThresholdByDatasetName(result.getDatasetName(), thresholdMap);
+ log.debug("Threshold is {} for dataset {}", threshold, result.getDatasetName());
InputRecordCountHelper helper = new InputRecordCountHelper(state);
try {
double newRecords = 0;
if (!dataset.isVirtual()) {
- newRecords = helper.calculateRecordCount (Lists.newArrayList(new Path(dataset.datasetURN())));
+ newRecords = helper.calculateRecordCount(Lists.newArrayList(new Path(dataset.datasetURN())));
}
- double oldRecords = helper.readRecordCount (new Path(result.getDstAbsoluteDir()));
+ double oldRecords = helper.readRecordCount(new Path(result.getDstAbsoluteDir()));
if (oldRecords == 0) {
return new Result(true, "");
}
+ if (newRecords < oldRecords) {
+ return new Result(false, "Illegal state: Current records count should old be smaller.");
+ }
+
if ((newRecords - oldRecords) / oldRecords > threshold) {
- log.debug ("Dataset {} records exceeded the threshold {}", dataset.datasetURN(), threshold);
+ log.debug("Dataset {} records exceeded the threshold {}", dataset.datasetURN(), threshold);
return new Result(true, "");
}
- return new Result(false, String.format("%s is failed for dataset %s. Prev=%f, Cur=%f, not reaching to threshold %f", this.getName(), result.getDatasetName(), oldRecords, newRecords, threshold));
+ return new Result(false, String
+ .format("%s is failed for dataset %s. Prev=%f, Cur=%f, not reaching to threshold %f", this.getName(),
+ result.getDatasetName(), oldRecords, newRecords, threshold));
} catch (IOException e) {
return new Result(false, ExceptionUtils.getFullStackTrace(e));
}
@@ -102,7 +109,7 @@ public class CompactionThresholdVerifier implements CompactionVerifier<FileSyste
return this.getClass().getName();
}
- public boolean isRetriable () {
+ public boolean isRetriable() {
return false;
}
}
diff --git a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/InputRecordCountHelper.java b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/InputRecordCountHelper.java
index e1bc952..bbc581d 100644
--- a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/InputRecordCountHelper.java
+++ b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/InputRecordCountHelper.java
@@ -43,8 +43,12 @@ import org.apache.gobblin.util.HadoopUtils;
import org.apache.gobblin.util.RecordCountProvider;
import org.apache.gobblin.util.recordcount.IngestionRecordCountProvider;
+import static org.apache.gobblin.compaction.mapreduce.CompactorOutputCommitter.COMPACTION_OUTPUT_EXTENSION;
+import static org.apache.gobblin.compaction.mapreduce.CompactorOutputCommitter.DEFAULT_COMPACTION_OUTPUT_EXTENSION;
+
+
/**
- * A class helps to calculate, serialize, deserialize record count.
+ * A class helps to calculate, serialize, deserialize record count. This will work for Avro and ORC formats.
*
* By using {@link IngestionRecordCountProvider}, the default input file name should be in format
* {file_name}.{record_count}.{extension}. For example, given a file path: "/a/b/c/file.123.avro",
@@ -57,7 +61,7 @@ public class InputRecordCountHelper {
private final FileSystem fs;
private final State state;
private final RecordCountProvider inputRecordCountProvider;
- private final String AVRO = "avro";
+ private final String extensionName;
@Deprecated
public final static String RECORD_COUNT_FILE = "_record_count";
@@ -71,6 +75,7 @@ public class InputRecordCountHelper {
try {
this.fs = getSourceFileSystem (state);
this.state = state;
+ this.extensionName = state.getProp(COMPACTION_OUTPUT_EXTENSION, DEFAULT_COMPACTION_OUTPUT_EXTENSION);
this.inputRecordCountProvider = (RecordCountProvider) Class
.forName(state.getProp(MRCompactor.COMPACTION_INPUT_RECORD_COUNT_PROVIDER,
MRCompactor.DEFAULT_COMPACTION_INPUT_RECORD_COUNT_PROVIDER))
@@ -88,7 +93,8 @@ public class InputRecordCountHelper {
public long calculateRecordCount (Collection<Path> paths) throws IOException {
long sum = 0;
for (Path path: paths) {
- sum += inputRecordCountProvider.getRecordCount(DatasetHelper.getApplicableFilePaths(this.fs, path, Lists.newArrayList(AVRO)));
+ sum += inputRecordCountProvider.getRecordCount(
+ DatasetHelper.getApplicableFilePaths(this.fs, path, Lists.newArrayList(extensionName)));
}
return sum;
}
diff --git a/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/AvroCompactionTaskTest.java b/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/AvroCompactionTaskTest.java
index 19d01fa..2a04d68 100644
--- a/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/AvroCompactionTaskTest.java
+++ b/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/AvroCompactionTaskTest.java
@@ -141,7 +141,7 @@ public class AvroCompactionTaskTest {
}
@Test
- public void testRecompaction () throws Exception {
+ public void testAvroRecompaction() throws Exception {
FileSystem fs = getFileSystem();
String basePath = "/tmp/testRecompaction";
fs.delete(new Path(basePath), true);
diff --git a/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/OrcCompactionTaskTest.java b/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/OrcCompactionTaskTest.java
index f855790..e6a3fab 100644
--- a/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/OrcCompactionTaskTest.java
+++ b/gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/OrcCompactionTaskTest.java
@@ -51,10 +51,13 @@ import org.testng.annotations.Test;
import static org.apache.gobblin.compaction.mapreduce.AvroCompactionTaskTest.*;
import static org.apache.gobblin.compaction.mapreduce.CompactorOutputCommitter.*;
+import static org.apache.gobblin.compaction.mapreduce.MRCompactor.COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET;
import static org.apache.gobblin.compaction.mapreduce.MRCompactor.COMPACTION_SHOULD_DEDUPLICATE;
public class OrcCompactionTaskTest {
+ final String extensionName = "orc";
+
private void createTestingData(File jobDir) throws Exception {
// Write some ORC file for compaction here.
TypeDescription schema = TypeDescription.fromString("struct<i:int,j:int>");
@@ -74,15 +77,16 @@ public class OrcCompactionTaskTest {
orcStruct_3.setFieldValue("i", new IntWritable(4));
orcStruct_3.setFieldValue("j", new IntWritable(5));
- File file_0 = new File(jobDir, "file_0");
- File file_1 = new File(jobDir, "file_1");
+ // Following pattern: FILENAME.RECORDCOUNT.EXTENSION
+ File file_0 = new File(jobDir, "file_0.2." + extensionName);
+ File file_1 = new File(jobDir, "file_1.2." + extensionName);
writeOrcRecordsInFile(new Path(file_0.getAbsolutePath()), schema, ImmutableList.of(orcStruct_0, orcStruct_2));
writeOrcRecordsInFile(new Path(file_1.getAbsolutePath()), schema, ImmutableList.of(orcStruct_1, orcStruct_3));
}
@Test
- public void basicTest() throws Exception {
+ public void basicTestWithRecompaction() throws Exception {
File basePath = Files.createTempDir();
basePath.deleteOnExit();
@@ -101,18 +105,18 @@ public class OrcCompactionTaskTest {
orcStruct_4.setFieldValue("j", new IntWritable(6));
orcStruct_4.setFieldValue("k", new IntWritable(7));
- File file_2 = new File(jobDir, "file_2");
+ File file_2 = new File(jobDir, "file_2.1." + extensionName);
writeOrcRecordsInFile(new Path(file_2.getAbsolutePath()), evolvedSchema, ImmutableList.of(orcStruct_4));
// Make this is the newest.
file_2.setLastModified(Long.MAX_VALUE);
// Verify execution
// Overwrite the job configurator factory key.
- String extensionFileName = "orcavro";
EmbeddedGobblin embeddedGobblin = createEmbeddedGobblin("basic", basePath.getAbsolutePath().toString())
.setConfiguration(CompactionJobConfigurator.COMPACTION_JOB_CONFIGURATOR_FACTORY_CLASS_KEY,
TestCompactionOrcJobConfigurator.Factory.class.getName())
- .setConfiguration(COMPACTION_OUTPUT_EXTENSION, extensionFileName);
+ .setConfiguration(COMPACTION_OUTPUT_EXTENSION, extensionName)
+ .setConfiguration(COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET, "Identity.*:0.1");
JobExecutionResult execution = embeddedGobblin.run();
Assert.assertTrue(execution.isSuccessful());
@@ -120,14 +124,7 @@ public class OrcCompactionTaskTest {
File outputDir = new File(basePath, hourlyPath);
FileSystem fs = FileSystem.getLocal(new Configuration());
List<FileStatus> statuses = new ArrayList<>();
- for (FileStatus status : fs.listStatus(new Path(outputDir.getAbsolutePath()), new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return FilenameUtils.isExtension(path.getName(), extensionFileName);
- }
- })) {
- statuses.add(status);
- }
+ reloadFolder(statuses, outputDir, fs);
Assert.assertTrue(statuses.size() == 1);
List<OrcStruct> result = readOrcFile(statuses.get(0).getPath());
@@ -144,6 +141,35 @@ public class OrcCompactionTaskTest {
Assert.assertEquals(result.get(3).getFieldValue("i"), new IntWritable(5));
Assert.assertEquals(result.get(3).getFieldValue("j"), new IntWritable(6));
Assert.assertEquals(result.get(3).getFieldValue("k"), new IntWritable(7));
+
+ // Adding new .orc file into the directory and verify if re-compaction is triggered.
+ File file_late = new File(jobDir, "file_late.1." + extensionName);
+ OrcStruct orcStruct_5 = (OrcStruct) OrcStruct.createValue(evolvedSchema);
+ orcStruct_5.setFieldValue("i", new IntWritable(10));
+ orcStruct_5.setFieldValue("j", new IntWritable(11));
+ orcStruct_5.setFieldValue("k", new IntWritable(12));
+
+ writeOrcRecordsInFile(new Path(file_late.getAbsolutePath()), evolvedSchema, ImmutableList.of(orcStruct_5));
+ execution = embeddedGobblin.run();
+ Assert.assertTrue(execution.isSuccessful());
+
+ reloadFolder(statuses, outputDir, fs);
+ result = readOrcFile(statuses.get(0).getPath());
+ // Note previous execution's inspection gives 4 result, given re-compaction, this should gives 1 late-record more.
+ Assert.assertEquals(result.size(), 4 + 1);
+ }
+
+ // A helper method to load all files in the output directory for compaction-result inspection.
+ private void reloadFolder(List<FileStatus> statuses, File outputDir, FileSystem fs) throws IOException {
+ statuses.clear();
+ for (FileStatus status : fs.listStatus(new Path(outputDir.getAbsolutePath()), new PathFilter() {
+ @Override
+ public boolean accept(Path path) {
+ return FilenameUtils.isExtension(path.getName(), extensionName);
+ }
+ })) {
+ statuses.add(status);
+ }
}
@Test