| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.druid.indexer; |
| |
| import com.fasterxml.jackson.core.JsonProcessingException; |
| import com.google.common.base.Predicate; |
| import com.google.common.base.Strings; |
| import org.apache.druid.java.util.common.DateTimes; |
| import org.apache.druid.java.util.common.FileUtils; |
| import org.apache.druid.java.util.common.IAE; |
| import org.apache.druid.java.util.common.IOE; |
| import org.apache.druid.java.util.common.ISE; |
| import org.apache.druid.java.util.common.RetryUtils; |
| import org.apache.druid.java.util.common.StringUtils; |
| import org.apache.druid.java.util.common.logger.Logger; |
| import org.apache.druid.segment.SegmentUtils; |
| import org.apache.druid.segment.loading.DataSegmentPusher; |
| import org.apache.druid.timeline.DataSegment; |
| import org.apache.druid.utils.CompressionUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.LocalFileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.io.retry.RetryPolicies; |
| import org.apache.hadoop.io.retry.RetryPolicy; |
| import org.apache.hadoop.io.retry.RetryProxy; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.mapreduce.MRJobConfig; |
| import org.apache.hadoop.mapreduce.TaskAttemptID; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.util.Progressable; |
| |
| import javax.annotation.Nullable; |
| import java.io.BufferedOutputStream; |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileNotFoundException; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.OutputStream; |
| import java.io.OutputStreamWriter; |
| import java.net.URI; |
| import java.net.URISyntaxException; |
| import java.nio.charset.StandardCharsets; |
| import java.nio.file.Files; |
| import java.nio.file.Paths; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.regex.Pattern; |
| import java.util.zip.ZipEntry; |
| import java.util.zip.ZipInputStream; |
| import java.util.zip.ZipOutputStream; |
| |
| /** |
| */ |
| public class JobHelper |
| { |
| private static final Logger log = new Logger(JobHelper.class); |
| private static final int NUM_RETRIES = 8; |
| private static final int SECONDS_BETWEEN_RETRIES = 2; |
| private static final int DEFAULT_FS_BUFFER_SIZE = 1 << 18; // 256KiB |
| private static final Pattern SNAPSHOT_JAR = Pattern.compile(".*-SNAPSHOT(-selfcontained)?\\.jar$"); |
| |
| public static Path distributedClassPath(String path) |
| { |
| return distributedClassPath(new Path(path)); |
| } |
| |
| public static Path distributedClassPath(Path base) |
| { |
| return new Path(base, "classpath"); |
| } |
| |
| public static final String INDEX_ZIP = "index.zip"; |
| |
| /** |
| * Dose authenticate against a secured hadoop cluster |
| * In case of any bug fix make sure to fix the code at HdfsStorageAuthentication#authenticate as well. |
| * |
| */ |
| public static void authenticate() |
| { |
| String principal = HadoopDruidIndexerConfig.HADOOP_KERBEROS_CONFIG.getPrincipal(); |
| String keytab = HadoopDruidIndexerConfig.HADOOP_KERBEROS_CONFIG.getKeytab(); |
| if (!Strings.isNullOrEmpty(principal) && !Strings.isNullOrEmpty(keytab)) { |
| Configuration conf = new Configuration(); |
| UserGroupInformation.setConfiguration(conf); |
| if (UserGroupInformation.isSecurityEnabled()) { |
| try { |
| if (UserGroupInformation.getCurrentUser().hasKerberosCredentials() == false |
| || !UserGroupInformation.getCurrentUser().getUserName().equals(principal)) { |
| log.info("trying to authenticate user [%s] with keytab [%s]", principal, keytab); |
| UserGroupInformation.loginUserFromKeytab(principal, keytab); |
| } |
| } |
| catch (IOException e) { |
| throw new ISE(e, "Failed to authenticate user principal [%s] with keytab [%s]", principal, keytab); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Uploads jar files to hdfs and configures the classpath. |
| * Snapshot jar files are uploaded to intermediateClasspath and not shared across multiple jobs. |
| * Non-Snapshot jar files are uploaded to a distributedClasspath and shared across multiple jobs. |
| * |
| * @param distributedClassPath classpath shared across multiple jobs |
| * @param intermediateClassPath classpath exclusive for this job. used to upload SNAPSHOT jar files. |
| * @param job job to run |
| * |
| * @throws IOException |
| */ |
| public static void setupClasspath( |
| final Path distributedClassPath, |
| final Path intermediateClassPath, |
| final Job job |
| ) |
| throws IOException |
| { |
| String classpathProperty = System.getProperty("druid.hadoop.internal.classpath"); |
| if (classpathProperty == null) { |
| classpathProperty = System.getProperty("java.class.path"); |
| } |
| |
| String[] jarFiles = classpathProperty.split(File.pathSeparator); |
| |
| final Configuration conf = job.getConfiguration(); |
| final FileSystem fs = distributedClassPath.getFileSystem(conf); |
| |
| if (fs instanceof LocalFileSystem) { |
| return; |
| } |
| |
| for (String jarFilePath : jarFiles) { |
| |
| final File jarFile = new File(jarFilePath); |
| if (jarFile.getName().endsWith(".jar")) { |
| try { |
| RetryUtils.retry( |
| () -> { |
| if (isSnapshot(jarFile)) { |
| addSnapshotJarToClassPath(jarFile, intermediateClassPath, fs, job); |
| } else { |
| addJarToClassPath(jarFile, distributedClassPath, intermediateClassPath, fs, job); |
| } |
| return true; |
| }, |
| shouldRetryPredicate(), |
| NUM_RETRIES |
| ); |
| } |
| catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| } |
| |
| public static Predicate<Throwable> shouldRetryPredicate() |
| { |
| return new Predicate<Throwable>() |
| { |
| @Override |
| public boolean apply(Throwable input) |
| { |
| if (input == null) { |
| return false; |
| } |
| if (input instanceof IOException) { |
| return true; |
| } |
| return apply(input.getCause()); |
| } |
| }; |
| } |
| |
| static void addJarToClassPath( |
| File jarFile, |
| Path distributedClassPath, |
| Path intermediateClassPath, |
| FileSystem fs, |
| Job job |
| ) |
| throws IOException |
| { |
| // Create distributed directory if it does not exist. |
| // rename will always fail if destination does not exist. |
| fs.mkdirs(distributedClassPath); |
| |
| // Non-snapshot jar files are uploaded to the shared classpath. |
| final Path hdfsPath = new Path(distributedClassPath, jarFile.getName()); |
| if (shouldUploadOrReplace(jarFile, hdfsPath, fs)) { |
| // Muliple jobs can try to upload the jar here, |
| // to avoid them from overwriting files, first upload to intermediateClassPath and then rename to the distributedClasspath. |
| final Path intermediateHdfsPath = new Path(intermediateClassPath, jarFile.getName()); |
| uploadJar(jarFile, intermediateHdfsPath, fs); |
| IOException exception = null; |
| try { |
| log.info("Renaming jar to path[%s]", hdfsPath); |
| fs.rename(intermediateHdfsPath, hdfsPath); |
| if (!fs.exists(hdfsPath)) { |
| throw new IOE("File does not exist even after moving from[%s] to [%s]", intermediateHdfsPath, hdfsPath); |
| } |
| } |
| catch (IOException e) { |
| // rename failed, possibly due to race condition. check if some other job has uploaded the jar file. |
| try { |
| if (!fs.exists(hdfsPath)) { |
| log.error(e, "IOException while Renaming jar file"); |
| exception = e; |
| } |
| } |
| catch (IOException e1) { |
| e.addSuppressed(e1); |
| exception = e; |
| } |
| } |
| finally { |
| try { |
| if (fs.exists(intermediateHdfsPath)) { |
| fs.delete(intermediateHdfsPath, false); |
| } |
| } |
| catch (IOException e) { |
| if (exception == null) { |
| exception = e; |
| } else { |
| exception.addSuppressed(e); |
| } |
| } |
| if (exception != null) { |
| throw exception; |
| } |
| } |
| } |
| job.addFileToClassPath(hdfsPath); |
| } |
| |
| static boolean shouldUploadOrReplace( |
| File jarFile, |
| Path hdfsPath, |
| FileSystem fs |
| ) |
| throws IOException |
| { |
| try { |
| FileStatus status = fs.getFileStatus(hdfsPath); |
| return status == null || status.getLen() != jarFile.length(); |
| } |
| catch (FileNotFoundException e) { |
| return true; |
| } |
| } |
| |
| static void addSnapshotJarToClassPath( |
| File jarFile, |
| Path intermediateClassPath, |
| FileSystem fs, |
| Job job |
| ) throws IOException |
| { |
| // Snapshot jars are uploaded to non shared intermediate directory. |
| final Path hdfsPath = new Path(intermediateClassPath, jarFile.getName()); |
| // Prevent uploading same file multiple times in same run. |
| if (!fs.exists(hdfsPath)) { |
| uploadJar(jarFile, hdfsPath, fs); |
| } |
| job.addFileToClassPath(hdfsPath); |
| } |
| |
| static void uploadJar(File jarFile, final Path path, final FileSystem fs) throws IOException |
| { |
| log.info("Uploading jar to path[%s]", path); |
| try (OutputStream os = fs.create(path)) { |
| Files.copy(jarFile.toPath(), os); |
| } |
| } |
| |
| static boolean isSnapshot(File jarFile) |
| { |
| return SNAPSHOT_JAR.matcher(jarFile.getName()).matches(); |
| } |
| |
| public static void injectDruidProperties(Configuration configuration, HadoopDruidIndexerConfig hadoopDruidIndexerConfig) |
| { |
| String mapJavaOpts = StringUtils.nullToEmptyNonDruidDataString(configuration.get(MRJobConfig.MAP_JAVA_OPTS)); |
| String reduceJavaOpts = StringUtils.nullToEmptyNonDruidDataString(configuration.get(MRJobConfig.REDUCE_JAVA_OPTS)); |
| |
| for (Map.Entry<String, String> allowedProperties : hadoopDruidIndexerConfig.getAllowedProperties().entrySet()) { |
| mapJavaOpts = StringUtils.format( |
| "%s -D%s=%s", |
| mapJavaOpts, |
| allowedProperties.getKey(), |
| allowedProperties.getValue() |
| ); |
| reduceJavaOpts = StringUtils.format( |
| "%s -D%s=%s", |
| reduceJavaOpts, |
| allowedProperties.getKey(), |
| allowedProperties.getValue() |
| ); |
| } |
| |
| if (!Strings.isNullOrEmpty(mapJavaOpts)) { |
| configuration.set(MRJobConfig.MAP_JAVA_OPTS, mapJavaOpts); |
| } |
| if (!Strings.isNullOrEmpty(reduceJavaOpts)) { |
| configuration.set(MRJobConfig.REDUCE_JAVA_OPTS, reduceJavaOpts); |
| } |
| } |
| |
| public static Configuration injectSystemProperties(Configuration conf, HadoopDruidIndexerConfig hadoopDruidIndexerConfig) |
| { |
| for (String propName : HadoopDruidIndexerConfig.PROPERTIES.stringPropertyNames()) { |
| if (propName.startsWith("hadoop.")) { |
| conf.set(propName.substring("hadoop.".length()), HadoopDruidIndexerConfig.PROPERTIES.getProperty(propName)); |
| } |
| } |
| |
| for (Map.Entry<String, String> allowedProperties : hadoopDruidIndexerConfig.getAllowedProperties().entrySet()) { |
| conf.set(allowedProperties.getKey(), allowedProperties.getValue()); |
| } |
| |
| return conf; |
| } |
| |
| public static void ensurePaths(HadoopDruidIndexerConfig config) |
| { |
| authenticate(); |
| // config.addInputPaths() can have side-effects ( boo! :( ), so this stuff needs to be done before anything else |
| try { |
| Job job = Job.getInstance( |
| new Configuration(), |
| StringUtils.format("%s-determine_partitions-%s", config.getDataSource(), config.getIntervals()) |
| ); |
| |
| job.getConfiguration().set("io.sort.record.percent", "0.19"); |
| injectSystemProperties(job.getConfiguration(), config); |
| config.addJobProperties(job); |
| |
| config.addInputPaths(job); |
| } |
| catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public static void writeJobIdToFile(String hadoopJobIdFileName, String hadoopJobId) |
| { |
| if (hadoopJobId != null && hadoopJobIdFileName != null) { |
| try (final OutputStream out = Files.newOutputStream(Paths.get(hadoopJobIdFileName))) { |
| try (final OutputStreamWriter osw = new OutputStreamWriter(out, StandardCharsets.UTF_8)) { |
| HadoopDruidIndexerConfig.JSON_MAPPER.writeValue( |
| osw, |
| hadoopJobId |
| ); |
| } |
| log.info("MR job id [%s] is written to the file [%s]", hadoopJobId, hadoopJobIdFileName); |
| } |
| catch (IOException e) { |
| log.warn(e, "Error writing job id [%s] to the file [%s]", hadoopJobId, hadoopJobIdFileName); |
| } |
| } else { |
| log.info("Either job id or file name is null for the submitted job. Skipping writing the file [%s]", hadoopJobIdFileName); |
| } |
| } |
| |
| public static boolean runSingleJob(Jobby job) |
| { |
| boolean succeeded = job.run(); |
| return succeeded; |
| } |
| |
| public static boolean runJobs(List<Jobby> jobs) |
| { |
| boolean succeeded = true; |
| for (Jobby job : jobs) { |
| if (!job.run()) { |
| succeeded = false; |
| break; |
| } |
| } |
| |
| return succeeded; |
| } |
| |
| public static void maybeDeleteIntermediatePath( |
| boolean jobSucceeded, |
| HadoopIngestionSpec indexerSchema) |
| { |
| HadoopDruidIndexerConfig config = HadoopDruidIndexerConfig.fromSpec(indexerSchema); |
| final Configuration configuration = JobHelper.injectSystemProperties(new Configuration(), config); |
| config.addJobProperties(configuration); |
| JobHelper.injectDruidProperties(configuration, config); |
| if (!config.getSchema().getTuningConfig().isLeaveIntermediate()) { |
| if (jobSucceeded || config.getSchema().getTuningConfig().isCleanupOnFailure()) { |
| Path workingPath = config.makeIntermediatePath(); |
| log.info("Deleting path[%s]", workingPath); |
| try { |
| config.addJobProperties(configuration); |
| workingPath.getFileSystem(configuration).delete(workingPath, true); |
| } |
| catch (IOException e) { |
| log.error(e, "Failed to cleanup path[%s]", workingPath); |
| } |
| } |
| } |
| } |
| |
| public static DataSegmentAndIndexZipFilePath serializeOutIndex( |
| final DataSegment segmentTemplate, |
| final Configuration configuration, |
| final Progressable progressable, |
| final File mergedBase, |
| final Path finalIndexZipFilePath, |
| final Path tmpPath, |
| DataSegmentPusher dataSegmentPusher |
| ) |
| throws IOException |
| { |
| final FileSystem outputFS = FileSystem.get(finalIndexZipFilePath.toUri(), configuration); |
| final AtomicLong size = new AtomicLong(0L); |
| final DataPusher zipPusher = (DataPusher) RetryProxy.create( |
| DataPusher.class, |
| new DataPusher() |
| { |
| @Override |
| public long push() throws IOException |
| { |
| try (OutputStream outputStream = outputFS.create( |
| tmpPath, |
| true, |
| DEFAULT_FS_BUFFER_SIZE, |
| progressable |
| )) { |
| size.set(zipAndCopyDir(mergedBase, outputStream, progressable)); |
| } |
| catch (IOException | RuntimeException exception) { |
| log.error(exception, "Exception in retry loop"); |
| throw exception; |
| } |
| return -1; |
| } |
| }, |
| RetryPolicies.exponentialBackoffRetry(NUM_RETRIES, SECONDS_BETWEEN_RETRIES, TimeUnit.SECONDS) |
| ); |
| zipPusher.push(); |
| log.info("Zipped %,d bytes to [%s]", size.get(), tmpPath.toUri()); |
| |
| final URI indexOutURI = finalIndexZipFilePath.toUri(); |
| final DataSegment finalSegment = segmentTemplate |
| .withLoadSpec(dataSegmentPusher.makeLoadSpec(indexOutURI)) |
| .withSize(size.get()) |
| .withBinaryVersion(SegmentUtils.getVersionFromDir(mergedBase)); |
| |
| return new DataSegmentAndIndexZipFilePath( |
| finalSegment, |
| tmpPath.toUri().toString(), |
| finalIndexZipFilePath.toUri().toString() |
| ); |
| } |
| |
| public static void writeSegmentDescriptor( |
| final FileSystem outputFS, |
| final DataSegmentAndIndexZipFilePath segmentAndPath, |
| final Path descriptorPath, |
| final Progressable progressable |
| ) |
| throws IOException |
| { |
| final DataPusher descriptorPusher = (DataPusher) RetryProxy.create( |
| DataPusher.class, |
| new DataPusher() |
| { |
| @Override |
| public long push() throws IOException |
| { |
| try { |
| progressable.progress(); |
| if (outputFS.exists(descriptorPath)) { |
| // If the descriptor path already exists, don't overwrite, and risk clobbering it. |
| // If it already exists, it means that the segment data is already written to the |
| // tmp path, and the existing descriptor written should give us the information we |
| // need to rename the segment index to final path and publish it in the top level task. |
| log.info("descriptor path [%s] already exists, not overwriting", descriptorPath); |
| return -1; |
| } |
| try (final OutputStream descriptorOut = outputFS.create( |
| descriptorPath, |
| true, |
| DEFAULT_FS_BUFFER_SIZE, |
| progressable |
| )) { |
| HadoopDruidIndexerConfig.JSON_MAPPER.writeValue(descriptorOut, segmentAndPath); |
| } |
| } |
| catch (RuntimeException | IOException ex) { |
| log.info(ex, "Exception in descriptor pusher retry loop"); |
| throw ex; |
| } |
| return -1; |
| } |
| }, |
| RetryPolicies.exponentialBackoffRetry(NUM_RETRIES, SECONDS_BETWEEN_RETRIES, TimeUnit.SECONDS) |
| ); |
| descriptorPusher.push(); |
| } |
| |
| /** |
| * Simple interface for retry operations |
| */ |
| public interface DataPusher |
| { |
| long push() throws IOException; |
| } |
| |
| public static long zipAndCopyDir( |
| File baseDir, |
| OutputStream baseOutputStream, |
| Progressable progressable |
| ) throws IOException |
| { |
| long size = 0L; |
| try (ZipOutputStream outputStream = new ZipOutputStream(baseOutputStream)) { |
| String[] filesToCopy = baseDir.list(); |
| if (filesToCopy != null) { |
| for (String fileName : filesToCopy) { |
| final File fileToCopy = new File(baseDir, fileName); |
| if (Files.isRegularFile(fileToCopy.toPath())) { |
| size += copyFileToZipStream(fileToCopy, outputStream, progressable); |
| } else { |
| log.warn("File at [%s] is not a regular file! skipping as part of zip", fileToCopy.getPath()); |
| } |
| } |
| } |
| outputStream.flush(); |
| } |
| return size; |
| } |
| |
| public static long copyFileToZipStream( |
| File file, |
| ZipOutputStream zipOutputStream, |
| Progressable progressable |
| ) throws IOException |
| { |
| createNewZipEntry(zipOutputStream, file); |
| long numRead = 0; |
| try (FileInputStream inputStream = new FileInputStream(file)) { |
| byte[] buf = new byte[0x10000]; |
| for (int bytesRead = inputStream.read(buf); bytesRead >= 0; bytesRead = inputStream.read(buf)) { |
| progressable.progress(); |
| if (bytesRead == 0) { |
| continue; |
| } |
| zipOutputStream.write(buf, 0, bytesRead); |
| progressable.progress(); |
| numRead += bytesRead; |
| } |
| } |
| zipOutputStream.closeEntry(); |
| progressable.progress(); |
| return numRead; |
| } |
| |
| private static void createNewZipEntry(ZipOutputStream out, File file) throws IOException |
| { |
| log.info("Creating new ZipEntry[%s]", file.getName()); |
| out.putNextEntry(new ZipEntry(file.getName())); |
| } |
| |
| public static Path makeFileNamePath( |
| final Path basePath, |
| final FileSystem fs, |
| final DataSegment segmentTemplate, |
| final String baseFileName, |
| DataSegmentPusher dataSegmentPusher |
| ) |
| { |
| return new Path( |
| prependFSIfNullScheme(fs, basePath), |
| dataSegmentPusher.makeIndexPathName(segmentTemplate, baseFileName) |
| ); |
| } |
| |
| public static Path makeTmpPath( |
| final Path basePath, |
| final FileSystem fs, |
| final DataSegment segmentTemplate, |
| final TaskAttemptID taskAttemptID, |
| DataSegmentPusher dataSegmentPusher |
| ) |
| { |
| return new Path( |
| prependFSIfNullScheme(fs, basePath), |
| StringUtils.format( |
| "./%s.%d", |
| dataSegmentPusher.makeIndexPathName(segmentTemplate, JobHelper.INDEX_ZIP), |
| taskAttemptID.getId() |
| ) |
| ); |
| } |
| |
| /** |
| * Renames the index files for the segments. This works around some limitations of both FileContext (no s3n support) and NativeS3FileSystem.rename |
| * which will not overwrite. Note: segments should be renamed in the index task, not in a hadoop job, as race |
| * conditions between job retries can cause the final segment index file path to get clobbered. |
| * |
| * @param indexerSchema the hadoop ingestion spec |
| * @param segmentAndIndexZipFilePaths the list of segments with their currently stored tmp path and the final path |
| * that they should be renamed to. |
| */ |
| public static void renameIndexFilesForSegments( |
| HadoopIngestionSpec indexerSchema, |
| List<DataSegmentAndIndexZipFilePath> segmentAndIndexZipFilePaths |
| ) throws IOException |
| { |
| HadoopDruidIndexerConfig config = HadoopDruidIndexerConfig.fromSpec(indexerSchema); |
| final Configuration configuration = JobHelper.injectSystemProperties(new Configuration(), config); |
| config.addJobProperties(configuration); |
| JobHelper.injectDruidProperties(configuration, config); |
| for (DataSegmentAndIndexZipFilePath segmentAndIndexZipFilePath : segmentAndIndexZipFilePaths) { |
| Path tmpPath = new Path(segmentAndIndexZipFilePath.getTmpIndexZipFilePath()); |
| Path finalIndexZipFilePath = new Path(segmentAndIndexZipFilePath.getFinalIndexZipFilePath()); |
| final FileSystem outputFS = FileSystem.get(finalIndexZipFilePath.toUri(), configuration); |
| if (!renameIndexFile(outputFS, tmpPath, finalIndexZipFilePath)) { |
| throw new IOE( |
| "Unable to rename [%s] to [%s]", |
| tmpPath.toUri().toString(), |
| finalIndexZipFilePath.toUri().toString() |
| ); |
| } |
| } |
| } |
| |
| /** |
| * Rename the file. This works around some limitations of both FileContext (no s3n support) and NativeS3FileSystem.rename |
| * which will not overwrite |
| * |
| * @param outputFS The output fs |
| * @param indexZipFilePath The original file path |
| * @param finalIndexZipFilePath The to rename the original file to |
| * |
| * @return False if a rename failed, true otherwise (rename success or no rename needed) |
| */ |
| private static boolean renameIndexFile( |
| final FileSystem outputFS, |
| final Path indexZipFilePath, |
| final Path finalIndexZipFilePath |
| ) |
| { |
| try { |
| return RetryUtils.retry( |
| () -> { |
| final boolean needRename; |
| |
| if (outputFS.exists(finalIndexZipFilePath)) { |
| // NativeS3FileSystem.rename won't overwrite, so we might need to delete the old index first |
| final FileStatus zipFile = outputFS.getFileStatus(indexZipFilePath); |
| final FileStatus finalIndexZipFile = outputFS.getFileStatus(finalIndexZipFilePath); |
| |
| if (zipFile.getModificationTime() >= finalIndexZipFile.getModificationTime() |
| || zipFile.getLen() != finalIndexZipFile.getLen()) { |
| log.info( |
| "File[%s / %s / %sB] existed, but wasn't the same as [%s / %s / %sB]", |
| finalIndexZipFile.getPath(), |
| DateTimes.utc(finalIndexZipFile.getModificationTime()), |
| finalIndexZipFile.getLen(), |
| zipFile.getPath(), |
| DateTimes.utc(zipFile.getModificationTime()), |
| zipFile.getLen() |
| ); |
| outputFS.delete(finalIndexZipFilePath, false); |
| needRename = true; |
| } else { |
| log.info( |
| "File[%s / %s / %sB] existed and will be kept", |
| finalIndexZipFile.getPath(), |
| DateTimes.utc(finalIndexZipFile.getModificationTime()), |
| finalIndexZipFile.getLen() |
| ); |
| needRename = false; |
| } |
| } else { |
| needRename = true; |
| } |
| |
| if (needRename) { |
| log.info("Attempting rename from [%s] to [%s]", indexZipFilePath, finalIndexZipFilePath); |
| return outputFS.rename(indexZipFilePath, finalIndexZipFilePath); |
| } else { |
| return true; |
| } |
| }, |
| FileUtils.IS_EXCEPTION, |
| NUM_RETRIES |
| ); |
| } |
| catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| |
| public static Path prependFSIfNullScheme(FileSystem fs, Path path) |
| { |
| if (path.toUri().getScheme() == null) { |
| path = fs.makeQualified(path); |
| } |
| return path; |
| } |
| |
| // TODO: Replace this whenever hadoop gets their act together and stops breaking with more recent versions of Guava |
| public static long unzipNoGuava( |
| final Path zip, |
| final Configuration configuration, |
| final File outDir, |
| final Progressable progressable, |
| @Nullable final RetryPolicy retryPolicy |
| ) throws IOException |
| { |
| final RetryPolicy effectiveRetryPolicy; |
| if (retryPolicy == null) { |
| effectiveRetryPolicy = RetryPolicies.exponentialBackoffRetry( |
| NUM_RETRIES, |
| SECONDS_BETWEEN_RETRIES, |
| TimeUnit.SECONDS |
| ); |
| } else { |
| effectiveRetryPolicy = retryPolicy; |
| } |
| |
| final DataPusher zipPusher = (DataPusher) RetryProxy.create( |
| DataPusher.class, |
| new DataPusher() |
| { |
| @Override |
| public long push() throws IOException |
| { |
| try { |
| final FileSystem fileSystem = zip.getFileSystem(configuration); |
| long size = 0L; |
| final byte[] buffer = new byte[1 << 13]; |
| progressable.progress(); |
| try (ZipInputStream in = new ZipInputStream(fileSystem.open(zip, 1 << 13))) { |
| for (ZipEntry entry = in.getNextEntry(); entry != null; entry = in.getNextEntry()) { |
| final String fileName = entry.getName(); |
| final String outputPath = new File(outDir, fileName).getAbsolutePath(); |
| |
| CompressionUtils.validateZipOutputFile(zip.getName(), new File(outputPath), outDir); |
| |
| try (final OutputStream out = new BufferedOutputStream(new FileOutputStream(outputPath))) { |
| for (int len = in.read(buffer); len >= 0; len = in.read(buffer)) { |
| progressable.progress(); |
| if (len == 0) { |
| continue; |
| } |
| size += len; |
| out.write(buffer, 0, len); |
| } |
| out.flush(); |
| } |
| } |
| } |
| progressable.progress(); |
| return size; |
| } |
| catch (IOException | RuntimeException exception) { |
| log.error(exception, "Exception in unzip retry loop"); |
| throw exception; |
| } |
| } |
| }, |
| effectiveRetryPolicy |
| ); |
| return zipPusher.push(); |
| } |
| |
| public static URI getURIFromSegment(DataSegment dataSegment) |
| { |
| // There is no good way around this... |
| // TODO: add getURI() to URIDataPuller |
| final Map<String, Object> loadSpec = dataSegment.getLoadSpec(); |
| final String type = loadSpec.get("type").toString(); |
| final URI segmentLocURI; |
| if ("s3_zip".equals(type)) { |
| if ("s3a".equals(loadSpec.get("S3Schema"))) { |
| segmentLocURI = URI.create(StringUtils.format("s3a://%s/%s", loadSpec.get("bucket"), loadSpec.get("key"))); |
| |
| } else { |
| segmentLocURI = URI.create(StringUtils.format("s3n://%s/%s", loadSpec.get("bucket"), loadSpec.get("key"))); |
| } |
| } else if ("hdfs".equals(type)) { |
| segmentLocURI = URI.create(loadSpec.get("path").toString()); |
| } else if ("google".equals(type)) { |
| // Segment names contain : in their path. |
| // Google Cloud Storage supports : but Hadoop does not. |
| // This becomes an issue when re-indexing using the current segments. |
| // The Hadoop getSplits code doesn't understand the : and returns "Relative path in absolute URI" |
| // This could be fixed using the same code that generates path names for hdfs segments using |
| // getHdfsStorageDir. But that wouldn't fix this issue for people who already have segments with ":". |
| // Because of this we just URL encode the : making everything work as it should. |
| segmentLocURI = URI.create( |
| StringUtils.format( |
| "gs://%s/%s", |
| loadSpec.get("bucket"), |
| StringUtils.replaceChar(loadSpec.get("path").toString(), ':', "%3A") |
| ) |
| ); |
| } else if ("local".equals(type)) { |
| try { |
| segmentLocURI = new URI("file", null, loadSpec.get("path").toString(), null, null); |
| } |
| catch (URISyntaxException e) { |
| throw new ISE(e, "Unable to form simple file uri"); |
| } |
| } else { |
| try { |
| throw new IAE( |
| "Cannot figure out loadSpec %s", |
| HadoopDruidIndexerConfig.JSON_MAPPER.writeValueAsString(loadSpec) |
| ); |
| } |
| catch (JsonProcessingException e) { |
| throw new ISE("Cannot write Map with json mapper"); |
| } |
| } |
| return segmentLocURI; |
| } |
| |
| public static String getJobTrackerAddress(Configuration config) |
| { |
| String jobTrackerAddress = config.get("mapred.job.tracker"); |
| if (jobTrackerAddress == null) { |
| // New Property name for Hadoop 3.0 and later versions |
| jobTrackerAddress = config.get("mapreduce.jobtracker.address"); |
| } |
| return jobTrackerAddress; |
| } |
| } |
