Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix exists check with compression

This commit is contained in:
nishantmonu51 2014-05-01 15:01:10 +05:30
parent 01e84f10b7
commit 728f1e8ee3
3 changed files with 20 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
indexing-hadoop/src/main/java/io/druid/indexer/DetermineHashedPartitionsJob.java
View File

@ -128,7 +128,7 @@ public class DetermineHashedPartitionsJob implements Jobby
if (!config.getSegmentGranularIntervals().isPresent()) { if (!config.getSegmentGranularIntervals().isPresent()) {
final Path intervalInfoPath = config.makeIntervalInfoPath(); final Path intervalInfoPath = config.makeIntervalInfoPath();
fileSystem = intervalInfoPath.getFileSystem(groupByJob.getConfiguration()); fileSystem = intervalInfoPath.getFileSystem(groupByJob.getConfiguration());
if (!fileSystem.exists(intervalInfoPath)) { if (!Utils.exists(groupByJob, fileSystem, intervalInfoPath)) {
throw new ISE("Path[%s] didn't exist!?", intervalInfoPath); throw new ISE("Path[%s] didn't exist!?", intervalInfoPath);
} }
List<Interval> intervals = config.jsonMapper.readValue( List<Interval> intervals = config.jsonMapper.readValue(
@ -148,7 +148,7 @@ public class DetermineHashedPartitionsJob implements Jobby
if (fileSystem == null) { if (fileSystem == null) {
fileSystem = partitionInfoPath.getFileSystem(groupByJob.getConfiguration()); fileSystem = partitionInfoPath.getFileSystem(groupByJob.getConfiguration());
} }
if (fileSystem.exists(partitionInfoPath)) { if (Utils.exists(groupByJob, fileSystem, partitionInfoPath)) {
Long cardinality = config.jsonMapper.readValue( Long cardinality = config.jsonMapper.readValue(
Utils.openInputStream(groupByJob, partitionInfoPath), new TypeReference<Long>() Utils.openInputStream(groupByJob, partitionInfoPath), new TypeReference<Long>()
{ {

2
indexing-hadoop/src/main/java/io/druid/indexer/DeterminePartitionsJob.java
View File

@ -215,7 +215,7 @@ public class DeterminePartitionsJob implements Jobby
if (fileSystem == null) { if (fileSystem == null) {
fileSystem = partitionInfoPath.getFileSystem(dimSelectionJob.getConfiguration()); fileSystem = partitionInfoPath.getFileSystem(dimSelectionJob.getConfiguration());
} }
if (fileSystem.exists(partitionInfoPath)) { if (Utils.exists(dimSelectionJob, fileSystem, partitionInfoPath)) {
List<ShardSpec> specs = config.jsonMapper.readValue( List<ShardSpec> specs = config.jsonMapper.readValue(
Utils.openInputStream(dimSelectionJob, partitionInfoPath), new TypeReference<List<ShardSpec>>() Utils.openInputStream(dimSelectionJob, partitionInfoPath), new TypeReference<List<ShardSpec>>()
{ {

21
indexing-hadoop/src/main/java/io/druid/indexer/Utils.java
View File

@ -65,7 +65,8 @@ public class Utils
return retVal; return retVal;
} }
public static OutputStream makePathAndOutputStream(JobContext job, Path outputPath, boolean deleteExisting) throws IOException public static OutputStream makePathAndOutputStream(JobContext job, Path outputPath, boolean deleteExisting)
throws IOException
{ {
OutputStream retVal; OutputStream retVal;
FileSystem fs = outputPath.getFileSystem(job.getConfiguration()); FileSystem fs = outputPath.getFileSystem(job.getConfiguration());
@ -73,8 +74,7 @@ public class Utils
if (fs.exists(outputPath)) { if (fs.exists(outputPath)) {
if (deleteExisting) { if (deleteExisting) {
fs.delete(outputPath, false); fs.delete(outputPath, false);
} } else {
else {
throw new ISE("outputPath[%s] must not exist.", outputPath); throw new ISE("outputPath[%s] must not exist.", outputPath);
} }
} }
@ -97,6 +97,17 @@ public class Utils
return openInputStream(inputPath, inputPath.getFileSystem(job.getConfiguration())); return openInputStream(inputPath, inputPath.getFileSystem(job.getConfiguration()));
} }
public static boolean exists(JobContext job, FileSystem fs, Path inputPath) throws IOException
{
if (!FileOutputFormat.getCompressOutput(job)) {
return fs.exists(inputPath);
} else {
Class<? extends CompressionCodec> codecClass = FileOutputFormat.getOutputCompressorClass(job, GzipCodec.class);
CompressionCodec codec = ReflectionUtils.newInstance(codecClass, job.getConfiguration());
return fs.exists(new Path(inputPath.toString() + codec.getDefaultExtension()));
}
}
public static InputStream openInputStream(Path inputPath, final FileSystem fileSystem) throws IOException public static InputStream openInputStream(Path inputPath, final FileSystem fileSystem) throws IOException
{ {
return fileSystem.open(inputPath); return fileSystem.open(inputPath);
@ -109,7 +120,9 @@ public class Utils
return jsonMapper.readValue( return jsonMapper.readValue(
fs.open(statsPath), fs.open(statsPath),
new TypeReference<Map<String, Object>>(){} new TypeReference<Map<String, Object>>()
{
}
); );
} }