org.apache.hadoop.fs.FileContext#getFileContext ( )源码实例Demo

private Configuration readLaunchConfiguration() throws IOException
{
  Path appPath = new Path(appContext.getApplicationPath());
  Path  configFilePath = new Path(appPath, LogicalPlan.LAUNCH_CONFIG_FILE_NAME);
  try {
    LOG.debug("Reading launch configuration file ");
    URI uri = appPath.toUri();
    Configuration config = new YarnConfiguration();
    fileContext = uri.getScheme() == null ? FileContext.getFileContext(config) : FileContext.getFileContext(uri, config);
    FSDataInputStream is = fileContext.open(configFilePath);
    config.addResource(is);
    LOG.debug("Read launch configuration");
    return config;
  } catch (FileNotFoundException ex) {
    LOG.warn("Configuration file not found {}", configFilePath);
    return new Configuration();
  }
}
 

private void testCopyPartialHelper(int dataSize, int offset, long size) throws IOException
  {
    FileUtils.deleteQuietly(new File("target/IOUtilsTest"));
    File file = new File("target/IOUtilsTest/testCopyPartial/input");
    createDataFile(file, dataSize);

    FileContext fileContext = FileContext.getFileContext();
    DataInputStream inputStream = fileContext.open(new Path(file.getAbsolutePath()));

    Path output = new Path("target/IOUtilsTest/testCopyPartial/output");
    DataOutputStream outputStream = fileContext.create(output, EnumSet
        .of(CreateFlag.CREATE, CreateFlag.OVERWRITE), Options.CreateOpts.CreateParent.createParent());

    if (offset == 0) {
      IOUtils.copyPartial(inputStream, size, outputStream);
    } else {
      IOUtils.copyPartial(inputStream, offset, size, outputStream);
    }

    outputStream.close();

    Assert.assertTrue("output exists", fileContext.util().exists(output));
    Assert.assertEquals("output size", size, fileContext.getFileStatus(output).getLen());
//    FileUtils.deleteQuietly(new File("target/IOUtilsTest"));
  }
 

private Configuration readLaunchConfiguration() throws IOException
{
  Path appPath = new Path(appContext.getApplicationPath());
  Path  configFilePath = new Path(appPath, LogicalPlan.LAUNCH_CONFIG_FILE_NAME);
  try {
    LOG.debug("Reading launch configuration file ");
    URI uri = appPath.toUri();
    Configuration config = new YarnConfiguration();
    fileContext = uri.getScheme() == null ? FileContext.getFileContext(config) : FileContext.getFileContext(uri, config);
    FSDataInputStream is = fileContext.open(configFilePath);
    config.addResource(is);
    LOG.debug("Read launch configuration");
    return config;
  } catch (FileNotFoundException ex) {
    LOG.warn("Configuration file not found {}", configFilePath);
    return new Configuration();
  }
}
 

/**
 * Same test as above, but for FileContext.
 */
@Test
public void testFileContextDoesntDnsResolveLogicalURI() throws Exception {
  FileSystem fs = HATestUtil.configureFailoverFs(cluster, conf);
  NameService spyNS = spyOnNameService();
  String logicalHost = fs.getUri().getHost();
  Configuration haClientConf = fs.getConf();
  
  FileContext fc = FileContext.getFileContext(haClientConf);
  Path root = new Path("/");
  fc.listStatus(root);
  fc.listStatus(fc.makeQualified(root));
  fc.getDefaultFileSystem().getCanonicalServiceName();

  // Ensure that the logical hostname was never resolved.
  Mockito.verify(spyNS, Mockito.never()).lookupAllHostAddr(Mockito.eq(logicalHost));
}
 

@Before
public void setUp() throws Exception {
  fcTarget = fc;
  fcTarget2 = fc2;
  targetTestRoot = fileContextTestHelper.getAbsoluteTestRootPath(fc);
  targetTestRoot2 = fileContextTestHelper.getAbsoluteTestRootPath(fc2);

  fcTarget.delete(targetTestRoot, true);
  fcTarget2.delete(targetTestRoot2, true);
  fcTarget.mkdir(targetTestRoot, new FsPermission((short) 0750), true);
  fcTarget2.mkdir(targetTestRoot2, new FsPermission((short) 0750), true);

  fsViewConf = ViewFileSystemTestSetup.createConfig();
  setupMountPoints();
  fcView = FileContext.getFileContext(FsConstants.VIEWFS_URI, fsViewConf);
}
 

public AvroFileInputStream(FileStatus status) throws IOException {
  pos = 0;
  buffer = new byte[0];
  GenericDatumReader<Object> reader = new GenericDatumReader<Object>();
  FileContext fc = FileContext.getFileContext(new Configuration());
  fileReader =
    DataFileReader.openReader(new AvroFSInput(fc, status.getPath()),reader);
  Schema schema = fileReader.getSchema();
  writer = new GenericDatumWriter<Object>(schema);
  output = new ByteArrayOutputStream();
  JsonGenerator generator =
    new JsonFactory().createJsonGenerator(output, JsonEncoding.UTF8);
  MinimalPrettyPrinter prettyPrinter = new MinimalPrettyPrinter();
  prettyPrinter.setRootValueSeparator(System.getProperty("line.separator"));
  generator.setPrettyPrinter(prettyPrinter);
  encoder = EncoderFactory.get().jsonEncoder(schema, generator);
}
 

private static LocationFactory createDefaultLocationFactory(Configuration configuration) {
  try {
    FileContext fc = FileContext.getFileContext(configuration);
    String basePath = fc.getHomeDirectory().toUri().getPath();
    return new FileContextLocationFactory(configuration, basePath);
  } catch (IOException e) {
    throw Throwables.propagate(e);
  }
}
 

public LocalContainerLauncher(AppContext context,
                              TaskUmbilicalProtocol umbilical) {
  super(LocalContainerLauncher.class.getName());
  this.context = context;
  this.umbilical = umbilical;
      // umbilical:  MRAppMaster creates (taskAttemptListener), passes to us
      // (TODO/FIXME:  pointless to use RPC to talk to self; should create
      // LocalTaskAttemptListener or similar:  implement umbilical protocol
      // but skip RPC stuff)

  try {
    curFC = FileContext.getFileContext(curDir.toURI());
  } catch (UnsupportedFileSystemException ufse) {
    LOG.error("Local filesystem " + curDir.toURI().toString()
              + " is unsupported?? (should never happen)");
  }

  // Save list of files/dirs that are supposed to be present so can delete
  // any extras created by one task before starting subsequent task.  Note
  // that there's no protection against deleted or renamed localization;
  // users who do that get what they deserve (and will have to disable
  // uberization in order to run correctly).
  File[] curLocalFiles = curDir.listFiles();
  localizedFiles = new HashSet<File>(curLocalFiles.length);
  for (int j = 0; j < curLocalFiles.length; ++j) {
    localizedFiles.add(curLocalFiles[j]);
  }

  // Relocalization note/future FIXME (per chrisdo, 20110315):  At moment,
  // full localization info is in AppSubmissionContext passed from client to
  // RM and then to NM for AM-container launch:  no difference between AM-
  // localization and MapTask- or ReduceTask-localization, so can assume all
  // OK.  Longer-term, will need to override uber-AM container-localization
  // request ("needed resources") with union of regular-AM-resources + task-
  // resources (and, if maps and reduces ever differ, then union of all three
  // types), OR will need localizer service/API that uber-AM can request
  // after running (e.g., "localizeForTask()" or "localizeForMapTask()").
}
 

@Override
@Before
public void setUp() throws Exception {
  // create the test root on local_fs
  fcTarget = FileContext.getLocalFSFileContext();
  super.setUp(); // this sets up conf (and fcView which we replace)
  
  // Now create a viewfs using a mount table called "default"
  // hence viewfs://default/
  schemeWithAuthority = 
    new URI(FsConstants.VIEWFS_SCHEME, "default", "/", null, null);
  fcView = FileContext.getFileContext(schemeWithAuthority, conf);  
}
 

@BeforeClass
public static void clusterSetupAtBeginning() throws IOException {
  cluster = new MiniDFSCluster.Builder(clusterConf)
      .nnTopology(MiniDFSNNTopology.simpleFederatedTopology(2))
      .numDataNodes(2)
      .build();
  cluster.waitClusterUp();

  fc = FileContext.getFileContext(cluster.getURI(0), clusterConf);
  fc2 = FileContext.getFileContext(cluster.getURI(1), clusterConf);
}
 

@Test
public void testDelete() throws IOException
{
  testLoad();
  testMeta.storageAgent.delete(1, 1);
  Path appPath = new Path(testMeta.applicationPath);
  FileContext fileContext = FileContext.getFileContext();
  Assert.assertTrue("operator 2 window 1", fileContext.util().exists(new Path(appPath + "/" + 2 + "/" + 1)));
  Assert.assertFalse("operator 1 window 1", fileContext.util().exists(new Path(appPath + "/" + 1 + "/" + 1)));
}
 

public LocalContainerLauncher(AppContext context,
                              TaskUmbilicalProtocol umbilical) {
  super(LocalContainerLauncher.class.getName());
  this.context = context;
  this.umbilical = umbilical;
      // umbilical:  MRAppMaster creates (taskAttemptListener), passes to us
      // (TODO/FIXME:  pointless to use RPC to talk to self; should create
      // LocalTaskAttemptListener or similar:  implement umbilical protocol
      // but skip RPC stuff)

  try {
    curFC = FileContext.getFileContext(curDir.toURI());
  } catch (UnsupportedFileSystemException ufse) {
    LOG.error("Local filesystem " + curDir.toURI().toString()
              + " is unsupported?? (should never happen)");
  }

  // Save list of files/dirs that are supposed to be present so can delete
  // any extras created by one task before starting subsequent task.  Note
  // that there's no protection against deleted or renamed localization;
  // users who do that get what they deserve (and will have to disable
  // uberization in order to run correctly).
  File[] curLocalFiles = curDir.listFiles();
  localizedFiles = new HashSet<File>(curLocalFiles.length);
  for (int j = 0; j < curLocalFiles.length; ++j) {
    localizedFiles.add(curLocalFiles[j]);
  }

  // Relocalization note/future FIXME (per chrisdo, 20110315):  At moment,
  // full localization info is in AppSubmissionContext passed from client to
  // RM and then to NM for AM-container launch:  no difference between AM-
  // localization and MapTask- or ReduceTask-localization, so can assume all
  // OK.  Longer-term, will need to override uber-AM container-localization
  // request ("needed resources") with union of regular-AM-resources + task-
  // resources (and, if maps and reduces ever differ, then union of all three
  // types), OR will need localizer service/API that uber-AM can request
  // after running (e.g., "localizeForTask()" or "localizeForMapTask()").
}
 

/**
 * Get default file system URI for the cluster (used to ensure consistency
 * of history done/staging locations) over different context
 *
 * @return Default file context
 */
private static FileContext getDefaultFileContext() {
  // If FS_DEFAULT_NAME_KEY was set solely by core-default.xml then we ignore
  // ignore it. This prevents defaulting history paths to file system specified
  // by core-default.xml which would not make sense in any case. For a test
  // case to exploit this functionality it should create core-site.xml
  FileContext fc = null;
  Configuration defaultConf = new Configuration();
  String[] sources;
  sources = defaultConf.getPropertySources(
      CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY);
  if (sources != null &&
      (!Arrays.asList(sources).contains("core-default.xml") ||
      sources.length > 1)) {
    try {
      fc = FileContext.getFileContext(defaultConf);
      LOG.info("Default file system [" +
                fc.getDefaultFileSystem().getUri() + "]");
    } catch (UnsupportedFileSystemException e) {
      LOG.error("Unable to create default file context [" +
          defaultConf.get(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY) +
          "]",
          e);
    }
  }
  else {
    LOG.info("Default file system is set solely " +
        "by core-default.xml therefore -  ignoring");
  }

  return fc;
}
 

public LogReader(Configuration conf, Path remoteAppLogFile)
    throws IOException {
  FileContext fileContext = FileContext.getFileContext(conf);
  this.fsDataIStream = fileContext.open(remoteAppLogFile);
  reader =
      new TFile.Reader(this.fsDataIStream, fileContext.getFileStatus(
          remoteAppLogFile).getLen(), conf);
  this.scanner = reader.createScanner();
}
 

public static Path getPreviousJobHistoryPath(
    Configuration conf, ApplicationAttemptId applicationAttemptId)
    throws IOException {
  String jobId =
      TypeConverter.fromYarn(applicationAttemptId.getApplicationId())
        .toString();
  String jobhistoryDir =
      JobHistoryUtils.getConfiguredHistoryStagingDirPrefix(conf, jobId);
  Path histDirPath = FileContext.getFileContext(conf).makeQualified(
          new Path(jobhistoryDir));
  FileContext fc = FileContext.getFileContext(histDirPath.toUri(), conf);
  return fc.makeQualified(JobHistoryUtils.getStagingJobHistoryFile(
      histDirPath,jobId, (applicationAttemptId.getAttemptId() - 1)));
}
 

public FileContextBasedFsStateStore(FileSystem fs, String storeRootDir, Class<T> stateClass)
    throws UnsupportedFileSystemException {
  super(fs, storeRootDir, stateClass);
  this.fc = FileContext.getFileContext(this.fs.getUri());
}
 

@Override
public void serviceInit(Configuration conf) throws Exception {
  conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_TEZ_FRAMEWORK_NAME);
  // blacklisting disabled to prevent scheduling issues
  conf.setBoolean(TezConfiguration.TEZ_AM_NODE_BLACKLISTING_ENABLED, false);
  if (conf.get(MRJobConfig.MR_AM_STAGING_DIR) == null) {
    conf.set(MRJobConfig.MR_AM_STAGING_DIR, new File(getTestWorkDir(),
        "apps_staging_dir" + Path.SEPARATOR).getAbsolutePath());
  }
  
  if (conf.get(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC) == null) {
    // nothing defined. set quick delete value
    conf.setLong(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, 0l);
  }
  
  File appJarLocalFile = new File(MiniTezCluster.APPJAR);

  if (!appJarLocalFile.exists()) {
    String message = "TezAppJar " + MiniTezCluster.APPJAR
        + " not found. Exiting.";
    LOG.info(message);
    throw new TezUncheckedException(message);
  }
  
  FileSystem fs = FileSystem.get(conf);
  Path testRootDir = fs.makeQualified(new Path("target", getName() + "-tmpDir"));
  Path appRemoteJar = new Path(testRootDir, "TezAppJar.jar");
  // Copy AppJar and make it public.
  Path appMasterJar = new Path(MiniTezCluster.APPJAR);
  fs.copyFromLocalFile(appMasterJar, appRemoteJar);
  fs.setPermission(appRemoteJar, new FsPermission("777"));

  conf.set(TezConfiguration.TEZ_LIB_URIS, appRemoteJar.toUri().toString());
  LOG.info("Set TEZ-LIB-URI to: " + conf.get(TezConfiguration.TEZ_LIB_URIS));

  // VMEM monitoring disabled, PMEM monitoring enabled.
  conf.setBoolean(YarnConfiguration.NM_PMEM_CHECK_ENABLED, false);
  conf.setBoolean(YarnConfiguration.NM_VMEM_CHECK_ENABLED, false);

  conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY,  "000");

  try {
    Path stagingPath = FileContext.getFileContext(conf).makeQualified(
        new Path(conf.get(MRJobConfig.MR_AM_STAGING_DIR)));
    /*
     * Re-configure the staging path on Windows if the file system is localFs.
     * We need to use a absolute path that contains the drive letter. The unit
     * test could run on a different drive than the AM. We can run into the
     * issue that job files are localized to the drive where the test runs on,
     * while the AM starts on a different drive and fails to find the job
     * metafiles. Using absolute path can avoid this ambiguity.
     */
    if (Path.WINDOWS) {
      if (LocalFileSystem.class.isInstance(stagingPath.getFileSystem(conf))) {
        conf.set(MRJobConfig.MR_AM_STAGING_DIR,
            new File(conf.get(MRJobConfig.MR_AM_STAGING_DIR))
                .getAbsolutePath());
      }
    }
    FileContext fc=FileContext.getFileContext(stagingPath.toUri(), conf);
    if (fc.util().exists(stagingPath)) {
      LOG.info(stagingPath + " exists! deleting...");
      fc.delete(stagingPath, true);
    }
    LOG.info("mkdir: " + stagingPath);
    fc.mkdir(stagingPath, null, true);

    //mkdir done directory as well
    String doneDir =
        JobHistoryUtils.getConfiguredHistoryServerDoneDirPrefix(conf);
    Path doneDirPath = fc.makeQualified(new Path(doneDir));
    fc.mkdir(doneDirPath, null, true);
  } catch (IOException e) {
    throw new TezUncheckedException("Could not create staging directory. ", e);
  }
  conf.set(MRConfig.MASTER_ADDRESS, "test");

  //configure the shuffle service in NM
  conf.setStrings(YarnConfiguration.NM_AUX_SERVICES,
      new String[] { ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID });
  conf.setClass(String.format(YarnConfiguration.NM_AUX_SERVICE_FMT,
      ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID), ShuffleHandler.class,
      Service.class);

  // Non-standard shuffle port
  conf.setInt(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY, 0);

  conf.setClass(YarnConfiguration.NM_CONTAINER_EXECUTOR,
      DefaultContainerExecutor.class, ContainerExecutor.class);

  // TestMRJobs is for testing non-uberized operation only; see TestUberAM
  // for corresponding uberized tests.
  conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
  super.serviceInit(conf);
}
 

@Override
public void serviceInit(Configuration conf) throws Exception {
  conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_FRAMEWORK_NAME);
  if (conf.get(MRJobConfig.MR_AM_STAGING_DIR) == null) {
    conf.set(MRJobConfig.MR_AM_STAGING_DIR, new File(getTestWorkDir(),
        "apps_staging_dir/").getAbsolutePath());
  }

  // By default, VMEM monitoring disabled, PMEM monitoring enabled.
  if (!conf.getBoolean(
      MRConfig.MAPREDUCE_MINICLUSTER_CONTROL_RESOURCE_MONITORING,
      MRConfig.DEFAULT_MAPREDUCE_MINICLUSTER_CONTROL_RESOURCE_MONITORING)) {
    conf.setBoolean(YarnConfiguration.NM_PMEM_CHECK_ENABLED, false);
    conf.setBoolean(YarnConfiguration.NM_VMEM_CHECK_ENABLED, false);
  }

  conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY,  "000");

  try {
    Path stagingPath = FileContext.getFileContext(conf).makeQualified(
        new Path(conf.get(MRJobConfig.MR_AM_STAGING_DIR)));
    /*
     * Re-configure the staging path on Windows if the file system is localFs.
     * We need to use a absolute path that contains the drive letter. The unit
     * test could run on a different drive than the AM. We can run into the
     * issue that job files are localized to the drive where the test runs on,
     * while the AM starts on a different drive and fails to find the job
     * metafiles. Using absolute path can avoid this ambiguity.
     */
    if (Path.WINDOWS) {
      if (LocalFileSystem.class.isInstance(stagingPath.getFileSystem(conf))) {
        conf.set(MRJobConfig.MR_AM_STAGING_DIR,
            new File(conf.get(MRJobConfig.MR_AM_STAGING_DIR))
                .getAbsolutePath());
      }
    }
    FileContext fc=FileContext.getFileContext(stagingPath.toUri(), conf);
    if (fc.util().exists(stagingPath)) {
      LOG.info(stagingPath + " exists! deleting...");
      fc.delete(stagingPath, true);
    }
    LOG.info("mkdir: " + stagingPath);
    //mkdir the staging directory so that right permissions are set while running as proxy user
    fc.mkdir(stagingPath, null, true);
    //mkdir done directory as well 
    String doneDir = JobHistoryUtils.getConfiguredHistoryServerDoneDirPrefix(conf);
    Path doneDirPath = fc.makeQualified(new Path(doneDir));
    fc.mkdir(doneDirPath, null, true);
  } catch (IOException e) {
    throw new YarnRuntimeException("Could not create staging directory. ", e);
  }
  conf.set(MRConfig.MASTER_ADDRESS, "test"); // The default is local because of
                                           // which shuffle doesn't happen
  //configure the shuffle service in NM
  conf.setStrings(YarnConfiguration.NM_AUX_SERVICES,
      new String[] { ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID });
  conf.setClass(String.format(YarnConfiguration.NM_AUX_SERVICE_FMT,
      ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID), ShuffleHandler.class,
      Service.class);

  // Non-standard shuffle port
  conf.setInt(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY, 0);

  conf.setClass(YarnConfiguration.NM_CONTAINER_EXECUTOR,
      DefaultContainerExecutor.class, ContainerExecutor.class);

  // TestMRJobs is for testing non-uberized operation only; see TestUberAM
  // for corresponding uberized tests.
  conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);

  super.serviceInit(conf);
}
 

@Test
public void testLogDeletion() throws IOException {
  File[] localLogDirs = getLocalLogDirFiles(this.getClass().getName(), 2);
  String localLogDirsString =
      localLogDirs[0].getAbsolutePath() + ","
          + localLogDirs[1].getAbsolutePath();

  conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDirsString);
  conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, false);
  conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS, 0l);

  dirsHandler.init(conf);

  NonAggregatingLogHandler rawLogHandler =
      new NonAggregatingLogHandler(dispatcher, mockDelService, dirsHandler,
          new NMNullStateStoreService());
  NonAggregatingLogHandler logHandler = spy(rawLogHandler);
  AbstractFileSystem spylfs =
      spy(FileContext.getLocalFSFileContext().getDefaultFileSystem());
  FileContext lfs = FileContext.getFileContext(spylfs, conf);
  doReturn(lfs).when(logHandler)
    .getLocalFileContext(isA(Configuration.class));
  FsPermission defaultPermission =
      FsPermission.getDirDefault().applyUMask(lfs.getUMask());
  final FileStatus fs =
      new FileStatus(0, true, 1, 0, System.currentTimeMillis(), 0,
        defaultPermission, "", "",
        new Path(localLogDirs[0].getAbsolutePath()));
  doReturn(fs).when(spylfs).getFileStatus(isA(Path.class));

  logHandler.init(conf);
  logHandler.start();

  logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null,
      ContainerLogsRetentionPolicy.ALL_CONTAINERS, null));

  logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0));

  logHandler.handle(new LogHandlerAppFinishedEvent(appId));

  Path[] localAppLogDirs = new Path[2];
  localAppLogDirs[0] =
      new Path(localLogDirs[0].getAbsolutePath(), appId.toString());
  localAppLogDirs[1] =
      new Path(localLogDirs[1].getAbsolutePath(), appId.toString());

  testDeletionServiceCall(mockDelService, user, 5000, localAppLogDirs);
  logHandler.close();
  for (int i = 0; i < localLogDirs.length; i++) {
    FileUtils.deleteDirectory(localLogDirs[i]);
  }
}
 

@Override
public void initialize(URI uri, Configuration conf) throws IOException {
  super.initialize(uri, conf);
  fc = FileContext.getFileContext(conf);
}