| /* |
| * 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.hcatalog.mapreduce; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.hive.common.FileUtils; |
| import org.apache.hadoop.hive.conf.HiveConf; |
| import org.apache.hadoop.hive.metastore.HiveMetaStoreClient; |
| import org.apache.hadoop.hive.metastore.Warehouse; |
| import org.apache.hadoop.hive.metastore.api.FieldSchema; |
| import org.apache.hadoop.hive.metastore.api.InvalidOperationException; |
| import org.apache.hadoop.hive.metastore.api.MetaException; |
| import org.apache.hadoop.hive.metastore.api.Partition; |
| import org.apache.hadoop.hive.metastore.api.StorageDescriptor; |
| import org.apache.hadoop.hive.metastore.api.Table; |
| import org.apache.hadoop.mapred.HCatMapRedUtil; |
| import org.apache.hadoop.mapred.JobConf; |
| import org.apache.hadoop.mapreduce.JobContext; |
| import org.apache.hadoop.mapreduce.JobStatus.State; |
| import org.apache.hadoop.mapreduce.TaskAttemptContext; |
| import org.apache.hadoop.security.AccessControlException; |
| import org.apache.hcatalog.common.ErrorType; |
| import org.apache.hcatalog.common.HCatConstants; |
| import org.apache.hcatalog.common.HCatException; |
| import org.apache.hcatalog.common.HCatUtil; |
| import org.apache.hcatalog.data.schema.HCatFieldSchema; |
| import org.apache.hcatalog.data.schema.HCatSchema; |
| import org.apache.hcatalog.data.schema.HCatSchemaUtils; |
| import org.apache.hcatalog.har.HarOutputCommitterPostProcessor; |
| import org.apache.thrift.TException; |
| |
| import java.io.IOException; |
| import java.net.URI; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| |
| /** |
| * Part of the FileOutput*Container classes |
| * See {@link FileOutputFormatContainer} for more information |
| */ |
| class FileOutputCommitterContainer extends OutputCommitterContainer { |
| |
| private final boolean dynamicPartitioningUsed; |
| private boolean partitionsDiscovered; |
| |
| private Map<String, Map<String, String>> partitionsDiscoveredByPath; |
| private Map<String, JobContext> contextDiscoveredByPath; |
| private final HCatStorageHandler cachedStorageHandler; |
| |
| HarOutputCommitterPostProcessor harProcessor = new HarOutputCommitterPostProcessor(); |
| |
| private String ptnRootLocation = null; |
| |
| private OutputJobInfo jobInfo = null; |
| |
| /** |
| * @param context current JobContext |
| * @param baseCommitter OutputCommitter to contain |
| * @throws IOException |
| */ |
| public FileOutputCommitterContainer(JobContext context, |
| org.apache.hadoop.mapred.OutputCommitter baseCommitter) throws IOException { |
| super(context, baseCommitter); |
| jobInfo = HCatOutputFormat.getJobInfo(context); |
| dynamicPartitioningUsed = jobInfo.isDynamicPartitioningUsed(); |
| |
| this.partitionsDiscovered = !dynamicPartitioningUsed; |
| cachedStorageHandler = HCatUtil.getStorageHandler(context.getConfiguration(),jobInfo.getTableInfo().getStorerInfo()); |
| } |
| |
| @Override |
| public void abortTask(TaskAttemptContext context) throws IOException { |
| if (!dynamicPartitioningUsed){ |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| getBaseOutputCommitter().abortTask(HCatMapRedUtil.createTaskAttemptContext(context)); |
| } |
| } |
| |
| @Override |
| public void commitTask(TaskAttemptContext context) throws IOException { |
| if (!dynamicPartitioningUsed){ |
| //TODO fix this hack, something wrong with pig |
| //running multiple storers in a single job, the real output dir got overwritten or something |
| //the location in OutputJobInfo is still correct so we'll use that |
| //TestHCatStorer.testMultiPartColsInData() used to fail without this |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| getBaseOutputCommitter().commitTask(HCatMapRedUtil.createTaskAttemptContext(context)); |
| } |
| } |
| |
| @Override |
| public boolean needsTaskCommit(TaskAttemptContext context) throws IOException { |
| if (!dynamicPartitioningUsed){ |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| return getBaseOutputCommitter().needsTaskCommit(HCatMapRedUtil.createTaskAttemptContext(context)); |
| }else{ |
| // called explicitly through FileRecordWriterContainer.close() if dynamic - return false by default |
| return false; |
| } |
| } |
| |
| @Override |
| public void setupJob(JobContext context) throws IOException { |
| if(getBaseOutputCommitter() != null && !dynamicPartitioningUsed) { |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| getBaseOutputCommitter().setupJob(HCatMapRedUtil.createJobContext(context)); |
| } |
| // in dynamic usecase, called through FileRecordWriterContainer |
| } |
| |
| @Override |
| public void setupTask(TaskAttemptContext context) throws IOException { |
| if (!dynamicPartitioningUsed){ |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| getBaseOutputCommitter().setupTask(HCatMapRedUtil.createTaskAttemptContext(context)); |
| } |
| } |
| |
| @Override |
| public void abortJob(JobContext jobContext, State state) throws IOException { |
| org.apache.hadoop.mapred.JobContext |
| mapRedJobContext = HCatMapRedUtil.createJobContext(jobContext); |
| if (dynamicPartitioningUsed){ |
| discoverPartitions(jobContext); |
| } |
| |
| if(getBaseOutputCommitter() != null && !dynamicPartitioningUsed) { |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(mapRedJobContext.getConfiguration()); |
| getBaseOutputCommitter().abortJob(mapRedJobContext, state); |
| } |
| else if (dynamicPartitioningUsed){ |
| for(JobContext currContext : contextDiscoveredByPath.values()){ |
| try { |
| new JobConf(currContext.getConfiguration()).getOutputCommitter().abortJob(currContext, state); |
| } catch (Exception e) { |
| throw new IOException(e); |
| } |
| } |
| } |
| |
| OutputJobInfo jobInfo = HCatOutputFormat.getJobInfo(jobContext); |
| |
| try { |
| HiveConf hiveConf = HCatUtil.getHiveConf(jobContext.getConfiguration()); |
| HiveMetaStoreClient client = HCatUtil.createHiveClient(hiveConf); |
| // cancel the deleg. tokens that were acquired for this job now that |
| // we are done - we should cancel if the tokens were acquired by |
| // HCatOutputFormat and not if they were supplied by Oozie. |
| // In the latter case the HCAT_KEY_TOKEN_SIGNATURE property in |
| // the conf will not be set |
| String tokenStrForm = client.getTokenStrForm(); |
| if(tokenStrForm != null && jobContext.getConfiguration().get |
| (HCatConstants.HCAT_KEY_TOKEN_SIGNATURE) != null) { |
| client.cancelDelegationToken(tokenStrForm); |
| } |
| |
| if (harProcessor.isEnabled()) { |
| String jcTokenStrForm = jobContext.getConfiguration().get(HCatConstants.HCAT_KEY_JOBCLIENT_TOKEN_STRFORM); |
| String jcTokenSignature = jobContext.getConfiguration().get(HCatConstants.HCAT_KEY_JOBCLIENT_TOKEN_SIGNATURE); |
| if(jcTokenStrForm != null && jcTokenSignature != null) { |
| HCatUtil.cancelJobTrackerDelegationToken(tokenStrForm,jcTokenSignature); |
| } |
| } |
| |
| } catch(Exception e) { |
| if( e instanceof HCatException ) { |
| throw (HCatException) e; |
| } else { |
| throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e); |
| } |
| } |
| |
| Path src; |
| if (dynamicPartitioningUsed){ |
| src = new Path(getPartitionRootLocation( |
| jobInfo.getLocation().toString(),jobInfo.getTableInfo().getTable().getPartitionKeysSize() |
| )); |
| }else{ |
| src = new Path(jobInfo.getLocation()); |
| } |
| FileSystem fs = src.getFileSystem(jobContext.getConfiguration()); |
| // LOG.warn("abortJob about to delete ["+src.toString() +"]"); |
| fs.delete(src, true); |
| } |
| |
| public static final String SUCCEEDED_FILE_NAME = "_SUCCESS"; |
| static final String SUCCESSFUL_JOB_OUTPUT_DIR_MARKER = |
| "mapreduce.fileoutputcommitter.marksuccessfuljobs"; |
| |
| private static boolean getOutputDirMarking(Configuration conf) { |
| return conf.getBoolean(SUCCESSFUL_JOB_OUTPUT_DIR_MARKER, |
| false); |
| } |
| |
| @Override |
| public void commitJob(JobContext jobContext) throws IOException { |
| if (dynamicPartitioningUsed){ |
| discoverPartitions(jobContext); |
| } |
| if(getBaseOutputCommitter() != null && !dynamicPartitioningUsed) { |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(jobContext.getConfiguration()); |
| getBaseOutputCommitter().commitJob(HCatMapRedUtil.createJobContext(jobContext)); |
| } |
| // create _SUCCESS FILE if so requested. |
| OutputJobInfo jobInfo = HCatOutputFormat.getJobInfo(jobContext); |
| if(getOutputDirMarking(jobContext.getConfiguration())) { |
| Path outputPath = new Path(jobInfo.getLocation()); |
| if (outputPath != null) { |
| FileSystem fileSys = outputPath.getFileSystem(jobContext.getConfiguration()); |
| // create a file in the folder to mark it |
| if (fileSys.exists(outputPath)) { |
| Path filePath = new Path(outputPath, SUCCEEDED_FILE_NAME); |
| if(!fileSys.exists(filePath)) { // may have been created by baseCommitter.commitJob() |
| fileSys.create(filePath).close(); |
| } |
| } |
| } |
| } |
| cleanupJob(jobContext); |
| } |
| |
| @Override |
| public void cleanupJob(JobContext context) throws IOException { |
| |
| if (dynamicPartitioningUsed){ |
| discoverPartitions(context); |
| } |
| |
| |
| OutputJobInfo jobInfo = HCatOutputFormat.getJobInfo(context); |
| Configuration conf = context.getConfiguration(); |
| Table table = jobInfo.getTableInfo().getTable(); |
| Path tblPath = new Path(table.getSd().getLocation()); |
| FileSystem fs = tblPath.getFileSystem(conf); |
| |
| if( table.getPartitionKeys().size() == 0 ) { |
| //non partitioned table |
| if(getBaseOutputCommitter() != null && !dynamicPartitioningUsed) { |
| // TODO: Hack! Pig messes up mapred.output.dir, when 2 Storers are used in the same Pig script. |
| // Workaround: Set mapred.output.dir from OutputJobInfo. |
| resetMapRedOutputDirFromJobInfo(context.getConfiguration()); |
| getBaseOutputCommitter().cleanupJob(HCatMapRedUtil.createJobContext(context)); |
| } |
| else if (dynamicPartitioningUsed){ |
| for(JobContext currContext : contextDiscoveredByPath.values()){ |
| try { |
| JobConf jobConf = new JobConf(currContext.getConfiguration()); |
| jobConf.getOutputCommitter().cleanupJob(currContext); |
| } catch (Exception e) { |
| throw new IOException(e); |
| } |
| } |
| } |
| |
| //Move data from temp directory the actual table directory |
| //No metastore operation required. |
| Path src = new Path(jobInfo.getLocation()); |
| moveTaskOutputs(fs, src, src, tblPath, false); |
| fs.delete(src, true); |
| return; |
| } |
| |
| HiveMetaStoreClient client = null; |
| HCatTableInfo tableInfo = jobInfo.getTableInfo(); |
| |
| List<Partition> partitionsAdded = new ArrayList<Partition>(); |
| |
| try { |
| HiveConf hiveConf = HCatUtil.getHiveConf(conf); |
| client = HCatUtil.createHiveClient(hiveConf); |
| |
| StorerInfo storer = InternalUtil.extractStorerInfo(table.getSd(),table.getParameters()); |
| |
| updateTableSchema(client, table, jobInfo.getOutputSchema()); |
| |
| FileStatus tblStat = fs.getFileStatus(tblPath); |
| String grpName = tblStat.getGroup(); |
| FsPermission perms = tblStat.getPermission(); |
| |
| List<Partition> partitionsToAdd = new ArrayList<Partition>(); |
| if (!dynamicPartitioningUsed){ |
| partitionsToAdd.add( |
| constructPartition( |
| context, |
| tblPath.toString(), jobInfo.getPartitionValues() |
| ,jobInfo.getOutputSchema(), getStorerParameterMap(storer) |
| ,table, fs |
| ,grpName,perms)); |
| }else{ |
| for (Entry<String,Map<String,String>> entry : partitionsDiscoveredByPath.entrySet()){ |
| partitionsToAdd.add( |
| constructPartition( |
| context, |
| getPartitionRootLocation(entry.getKey(),entry.getValue().size()), entry.getValue() |
| ,jobInfo.getOutputSchema(), getStorerParameterMap(storer) |
| ,table, fs |
| ,grpName,perms)); |
| } |
| } |
| |
| //Publish the new partition(s) |
| if (dynamicPartitioningUsed && harProcessor.isEnabled() && (!partitionsToAdd.isEmpty())){ |
| |
| Path src = new Path(ptnRootLocation); |
| |
| // check here for each dir we're copying out, to see if it already exists, error out if so |
| moveTaskOutputs(fs, src, src, tblPath,true); |
| |
| moveTaskOutputs(fs, src, src, tblPath,false); |
| fs.delete(src, true); |
| |
| |
| // for (Partition partition : partitionsToAdd){ |
| // partitionsAdded.add(client.add_partition(partition)); |
| // // currently following add_partition instead of add_partitions because latter isn't |
| // // all-or-nothing and we want to be able to roll back partitions we added if need be. |
| // } |
| |
| try { |
| client.add_partitions(partitionsToAdd); |
| partitionsAdded = partitionsToAdd; |
| } catch (Exception e){ |
| // There was an error adding partitions : rollback fs copy and rethrow |
| for (Partition p : partitionsToAdd){ |
| Path ptnPath = new Path(harProcessor.getParentFSPath(new Path(p.getSd().getLocation()))); |
| if (fs.exists(ptnPath)){ |
| fs.delete(ptnPath,true); |
| } |
| } |
| throw e; |
| } |
| |
| }else{ |
| // no harProcessor, regular operation |
| |
| // No duplicate partition publish case to worry about because we'll |
| // get a AlreadyExistsException here if so, and appropriately rollback |
| |
| client.add_partitions(partitionsToAdd); |
| partitionsAdded = partitionsToAdd; |
| |
| if (dynamicPartitioningUsed && (partitionsAdded.size()>0)){ |
| Path src = new Path(ptnRootLocation); |
| moveTaskOutputs(fs, src, src, tblPath,false); |
| fs.delete(src, true); |
| } |
| |
| } |
| |
| if(getBaseOutputCommitter() != null && !dynamicPartitioningUsed) { |
| getBaseOutputCommitter().cleanupJob(HCatMapRedUtil.createJobContext(context)); |
| } |
| |
| //Cancel HCat and JobTracker tokens |
| // cancel the deleg. tokens that were acquired for this job now that |
| // we are done - we should cancel if the tokens were acquired by |
| // HCatOutputFormat and not if they were supplied by Oozie. In the latter |
| // case the HCAT_KEY_TOKEN_SIGNATURE property in the conf will not be set |
| String tokenStrForm = client.getTokenStrForm(); |
| if(tokenStrForm != null && context.getConfiguration().get(HCatConstants.HCAT_KEY_TOKEN_SIGNATURE) != null) { |
| client.cancelDelegationToken(tokenStrForm); |
| } |
| if(harProcessor.isEnabled()) { |
| String jcTokenStrForm = |
| context.getConfiguration().get(HCatConstants.HCAT_KEY_JOBCLIENT_TOKEN_STRFORM); |
| String jcTokenSignature = |
| context.getConfiguration().get(HCatConstants.HCAT_KEY_JOBCLIENT_TOKEN_SIGNATURE); |
| if(jcTokenStrForm != null && jcTokenSignature != null) { |
| HCatUtil.cancelJobTrackerDelegationToken(tokenStrForm,jcTokenSignature); |
| } |
| } |
| } catch (Exception e) { |
| |
| if( partitionsAdded.size() > 0 ) { |
| try { |
| //baseCommitter.cleanupJob failed, try to clean up the metastore |
| for (Partition p : partitionsAdded){ |
| client.dropPartition(tableInfo.getDatabaseName(), |
| tableInfo.getTableName(), p.getValues()); |
| } |
| } catch(Exception te) { |
| //Keep cause as the original exception |
| throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e); |
| } |
| } |
| |
| if( e instanceof HCatException ) { |
| throw (HCatException) e; |
| } else { |
| throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e); |
| } |
| } finally { |
| if( client != null ) { |
| } |
| } |
| } |
| |
| private String getPartitionRootLocation(String ptnLocn,int numPtnKeys) { |
| if (ptnRootLocation == null){ |
| // we only need to calculate it once, it'll be the same for other partitions in this job. |
| Path ptnRoot = new Path(ptnLocn); |
| for (int i = 0; i < numPtnKeys; i++){ |
| // LOG.info("Getting parent of "+ptnRoot.getName()); |
| ptnRoot = ptnRoot.getParent(); |
| } |
| ptnRootLocation = ptnRoot.toString(); |
| } |
| // LOG.info("Returning final parent : "+ptnRootLocation); |
| return ptnRootLocation; |
| } |
| |
| /** |
| * Generate partition metadata object to be used to add to metadata. |
| * @param partLocnRoot The table-equivalent location root of the partition |
| * (temporary dir if dynamic partition, table dir if static) |
| * @param partKVs The keyvalue pairs that form the partition |
| * @param outputSchema The output schema for the partition |
| * @param params The parameters to store inside the partition |
| * @param table The Table metadata object under which this Partition will reside |
| * @param fs FileSystem object to operate on the underlying filesystem |
| * @param grpName Group name that owns the table dir |
| * @param perms FsPermission that's the default permission of the table dir. |
| * @return Constructed Partition metadata object |
| * @throws java.io.IOException |
| */ |
| |
| private Partition constructPartition( |
| JobContext context, |
| String partLocnRoot, Map<String,String> partKVs, |
| HCatSchema outputSchema, Map<String, String> params, |
| Table table, FileSystem fs, |
| String grpName, FsPermission perms) throws IOException { |
| |
| StorageDescriptor tblSD = table.getSd(); |
| |
| Partition partition = new Partition(); |
| partition.setDbName(table.getDbName()); |
| partition.setTableName(table.getTableName()); |
| partition.setSd(new StorageDescriptor(tblSD)); |
| |
| List<FieldSchema> fields = new ArrayList<FieldSchema>(); |
| for(HCatFieldSchema fieldSchema : outputSchema.getFields()) { |
| fields.add(HCatSchemaUtils.getFieldSchema(fieldSchema)); |
| } |
| |
| partition.getSd().setCols(fields); |
| |
| partition.setValues(FileOutputFormatContainer.getPartitionValueList(table, partKVs)); |
| |
| partition.setParameters(params); |
| |
| // Sets permissions and group name on partition dirs. |
| |
| Path partPath = new Path(partLocnRoot); |
| for(FieldSchema partKey : table.getPartitionKeys()){ |
| partPath = constructPartialPartPath(partPath, partKey.getName().toLowerCase(), partKVs); |
| // LOG.info("Setting perms for "+partPath.toString()); |
| fs.setPermission(partPath, perms); |
| try{ |
| fs.setOwner(partPath, null, grpName); |
| } catch(AccessControlException ace){ |
| // log the messages before ignoring. Currently, logging is not built in Hcatalog. |
| // LOG.warn(ace); |
| } |
| } |
| if (dynamicPartitioningUsed){ |
| String dynamicPartitionDestination = getFinalDynamicPartitionDestination(table,partKVs); |
| if (harProcessor.isEnabled()){ |
| harProcessor.exec(context, partition, partPath); |
| partition.getSd().setLocation( |
| harProcessor.getProcessedLocation(new Path(dynamicPartitionDestination))); |
| }else{ |
| partition.getSd().setLocation(dynamicPartitionDestination); |
| } |
| }else{ |
| partition.getSd().setLocation(partPath.toString()); |
| } |
| |
| return partition; |
| } |
| |
| |
| |
| private String getFinalDynamicPartitionDestination(Table table, Map<String,String> partKVs) { |
| // file:///tmp/hcat_junit_warehouse/employee/_DYN0.7770480401313761/emp_country=IN/emp_state=KA -> |
| // file:///tmp/hcat_junit_warehouse/employee/emp_country=IN/emp_state=KA |
| Path partPath = new Path(table.getSd().getLocation()); |
| for(FieldSchema partKey : table.getPartitionKeys()){ |
| partPath = constructPartialPartPath(partPath, partKey.getName().toLowerCase(), partKVs); |
| } |
| return partPath.toString(); |
| } |
| |
| private Map<String, String> getStorerParameterMap(StorerInfo storer) { |
| Map<String, String> params = new HashMap<String, String>(); |
| |
| //Copy table level hcat.* keys to the partition |
| for(Entry<Object, Object> entry : storer.getProperties().entrySet()) { |
| params.put(entry.getKey().toString(), entry.getValue().toString()); |
| } |
| return params; |
| } |
| |
| private Path constructPartialPartPath(Path partialPath, String partKey, Map<String,String> partKVs){ |
| |
| StringBuilder sb = new StringBuilder(FileUtils.escapePathName(partKey)); |
| sb.append("="); |
| sb.append(FileUtils.escapePathName(partKVs.get(partKey))); |
| return new Path(partialPath, sb.toString()); |
| } |
| |
| /** |
| * Update table schema, adding new columns as added for the partition. |
| * @param client the client |
| * @param table the table |
| * @param partitionSchema the schema of the partition |
| * @throws java.io.IOException Signals that an I/O exception has occurred. |
| * @throws org.apache.hadoop.hive.metastore.api.InvalidOperationException the invalid operation exception |
| * @throws org.apache.hadoop.hive.metastore.api.MetaException the meta exception |
| * @throws org.apache.thrift.TException the t exception |
| */ |
| private void updateTableSchema(HiveMetaStoreClient client, Table table, |
| HCatSchema partitionSchema) throws IOException, InvalidOperationException, MetaException, TException { |
| |
| List<FieldSchema> newColumns = HCatUtil.validatePartitionSchema(table, partitionSchema); |
| |
| if( newColumns.size() != 0 ) { |
| List<FieldSchema> tableColumns = new ArrayList<FieldSchema>(table.getSd().getCols()); |
| tableColumns.addAll(newColumns); |
| |
| //Update table schema to add the newly added columns |
| table.getSd().setCols(tableColumns); |
| client.alter_table(table.getDbName(), table.getTableName(), table); |
| } |
| } |
| |
| /** |
| * Move all of the files from the temp directory to the final location |
| * @param fs the output file system |
| * @param file the file to move |
| * @param src the source directory |
| * @param dest the target directory |
| * @param dryRun - a flag that simply tests if this move would succeed or not based |
| * on whether other files exist where we're trying to copy |
| * @throws java.io.IOException |
| */ |
| private void moveTaskOutputs(FileSystem fs, |
| Path file, |
| Path src, |
| Path dest, boolean dryRun) throws IOException { |
| if (fs.isFile(file)) { |
| Path finalOutputPath = getFinalPath(file, src, dest); |
| |
| if (dryRun){ |
| // LOG.info("Testing if moving ["+file+"] to ["+finalOutputPath+"] would cause a problem"); |
| if (fs.exists(finalOutputPath)){ |
| throw new HCatException(ErrorType.ERROR_MOVE_FAILED, "Data already exists in " + finalOutputPath + ", duplicate publish possible."); |
| } |
| }else{ |
| // LOG.info("Moving ["+file+"] to ["+finalOutputPath+"]"); |
| if (!fs.rename(file, finalOutputPath)) { |
| if (!fs.delete(finalOutputPath, true)) { |
| throw new HCatException(ErrorType.ERROR_MOVE_FAILED, "Failed to delete existing path " + finalOutputPath); |
| } |
| if (!fs.rename(file, finalOutputPath)) { |
| throw new HCatException(ErrorType.ERROR_MOVE_FAILED, "Failed to move output to " + dest); |
| } |
| } |
| } |
| } else if(fs.getFileStatus(file).isDir()) { |
| FileStatus[] paths = fs.listStatus(file); |
| Path finalOutputPath = getFinalPath(file, src, dest); |
| if (!dryRun){ |
| fs.mkdirs(finalOutputPath); |
| } |
| if (paths != null) { |
| for (FileStatus path : paths) { |
| moveTaskOutputs(fs, path.getPath(), src, dest,dryRun); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Find the final name of a given output file, given the output directory |
| * and the work directory. |
| * @param file the file to move |
| * @param src the source directory |
| * @param dest the target directory |
| * @return the final path for the specific output file |
| * @throws java.io.IOException |
| */ |
| private Path getFinalPath(Path file, Path src, |
| Path dest) throws IOException { |
| URI taskOutputUri = file.toUri(); |
| URI relativePath = src.toUri().relativize(taskOutputUri); |
| if (taskOutputUri == relativePath) { |
| throw new HCatException(ErrorType.ERROR_MOVE_FAILED, "Can not get the relative path: base = " + |
| src + " child = " + file); |
| } |
| if (relativePath.getPath().length() > 0) { |
| return new Path(dest, relativePath.getPath()); |
| } else { |
| return dest; |
| } |
| } |
| |
| /** |
| * Run to discover dynamic partitions available |
| */ |
| private void discoverPartitions(JobContext context) throws IOException { |
| if (!partitionsDiscovered){ |
| // LOG.info("discover ptns called"); |
| OutputJobInfo jobInfo = HCatOutputFormat.getJobInfo(context); |
| |
| harProcessor.setEnabled(jobInfo.getHarRequested()); |
| |
| List<Integer> dynamicPartCols = jobInfo.getPosOfDynPartCols(); |
| int maxDynamicPartitions = jobInfo.getMaxDynamicPartitions(); |
| |
| Path loadPath = new Path(jobInfo.getLocation()); |
| FileSystem fs = loadPath.getFileSystem(context.getConfiguration()); |
| |
| // construct a path pattern (e.g., /*/*) to find all dynamically generated paths |
| String dynPathSpec = loadPath.toUri().getPath(); |
| dynPathSpec = dynPathSpec.replaceAll("__HIVE_DEFAULT_PARTITION__", "*"); |
| |
| // LOG.info("Searching for "+dynPathSpec); |
| Path pathPattern = new Path(dynPathSpec); |
| FileStatus[] status = fs.globStatus(pathPattern); |
| |
| partitionsDiscoveredByPath = new LinkedHashMap<String,Map<String, String>>(); |
| contextDiscoveredByPath = new LinkedHashMap<String,JobContext>(); |
| |
| |
| if (status.length == 0) { |
| // LOG.warn("No partition found genereated by dynamic partitioning in [" |
| // +loadPath+"] with depth["+jobInfo.getTable().getPartitionKeysSize() |
| // +"], dynSpec["+dynPathSpec+"]"); |
| }else{ |
| if ((maxDynamicPartitions != -1) && (status.length > maxDynamicPartitions)){ |
| this.partitionsDiscovered = true; |
| throw new HCatException(ErrorType.ERROR_TOO_MANY_DYNAMIC_PTNS, |
| "Number of dynamic partitions being created " |
| + "exceeds configured max allowable partitions[" |
| + maxDynamicPartitions |
| + "], increase parameter [" |
| + HiveConf.ConfVars.DYNAMICPARTITIONMAXPARTS.varname |
| + "] if needed."); |
| } |
| |
| for (FileStatus st : status){ |
| LinkedHashMap<String, String> fullPartSpec = new LinkedHashMap<String, String>(); |
| Warehouse.makeSpecFromName(fullPartSpec, st.getPath()); |
| partitionsDiscoveredByPath.put(st.getPath().toString(),fullPartSpec); |
| JobContext currContext = new JobContext(context.getConfiguration(),context.getJobID()); |
| HCatOutputFormat.configureOutputStorageHandler(context, jobInfo, fullPartSpec); |
| contextDiscoveredByPath.put(st.getPath().toString(),currContext); |
| } |
| } |
| |
| // for (Entry<String,Map<String,String>> spec : partitionsDiscoveredByPath.entrySet()){ |
| // LOG.info("Partition "+ spec.getKey()); |
| // for (Entry<String,String> e : spec.getValue().entrySet()){ |
| // LOG.info(e.getKey() + "=>" +e.getValue()); |
| // } |
| // } |
| |
| this.partitionsDiscovered = true; |
| } |
| } |
| |
| /** |
| * TODO: Clean up this Hack! Resetting mapred.output.dir from OutputJobInfo. |
| * This works around PIG-2578, where Pig messes up output-directory |
| * if multiple storers are used in the same pig-script. |
| * @param config The configuration whose mapred.output.dir is to be reset. |
| */ |
| private void resetMapRedOutputDirFromJobInfo(Configuration config) { |
| String outputLocation = jobInfo.getLocation(); |
| if (outputLocation != null) |
| config.set("mapred.output.dir", outputLocation); |
| } |
| } |
