tensorflow_io/core/filesystems/hdfs/hadoop_filesystem.cc
// We implement connection caching in Tensorflow, which can significantly
// improve performance. Fixes #43187
We implement connection caching in Tensorflow, which can significantly
improve performance.
We implement connection caching in Tensorflow which can significantly improve performance .
hdfsFS Connect(tf_hdfs_filesystem::HadoopFileSystemImplementation* hadoop_file,
               const std::string& path, TF_Status* status) {
  auto libhdfs = hadoop_file->libhdfs;
  std::string scheme, namenode, hdfs_path;
  ParseHadoopPath(path, &scheme, &namenode, &hdfs_path);
  std::string cacheKey(scheme);
  if (scheme == "file") {
    namenode = "";
  } else if (scheme == "viewfs") {
    char* defaultFS = nullptr;
    libhdfs->hdfsConfGetStr("fs.defaultFS", &defaultFS);
    std::string defaultScheme, defaultCluster, defaultPath;
    ParseHadoopPath(defaultFS, &defaultScheme, &defaultCluster, &defaultPath);
    if (scheme != defaultScheme ||
        (namenode.empty() && namenode != defaultCluster)) {
      TF_SetStatus(status, TF_UNIMPLEMENTED,
                   "viewfs is only supported as a fs.defaultFS.");
      return nullptr;
    }
    namenode = "default";
  } else if (scheme == "har") {
    std::string path_har = path;
    SplitArchiveNameAndPath(&path_har, &namenode, status);
    if (TF_GetCode(status) != TF_OK) return nullptr;
  } else {
    if (namenode.empty()) {
      namenode = "default";
    }
  }
  cacheKey += namenode;
  absl::MutexLock l(&hadoop_file->connection_cache_lock);
  if (hadoop_file->connection_cache.find(cacheKey) ==
      hadoop_file->connection_cache.end()) {
    hdfsBuilder* builder = libhdfs->hdfsNewBuilder();
    libhdfs->hdfsBuilderSetNameNode(
        builder, namenode.empty() ? nullptr : namenode.c_str());
    auto cacheFs = libhdfs->hdfsBuilderConnect(builder);
    if (cacheFs == nullptr) {
      TF_SetStatusFromIOError(status, TF_ABORTED, strerror(errno));
      return cacheFs;
    }
    hadoop_file->connection_cache[cacheKey] = cacheFs;
  }
  auto fs = hadoop_file->connection_cache[cacheKey];
  TF_SetStatus(status, TF_OK, "");
  return fs;
}
hdfsFS Connect ( tf_hdfs_filesystem :: HadoopFileSystemImplementation * hadoop_file , const std :: string & path , TF_Status * status ) { auto libhdfs = hadoop_file -> libhdfs ; std :: string scheme , namenode , hdfs_path ; ParseHadoopPath ( path , & scheme , & namenode , & hdfs_path ) ; std :: string cacheKey ( scheme ) ; if ( scheme == " " ) { namenode = " " ; } else if ( scheme == " " ) { char * defaultFS = nullptr ; libhdfs -> hdfsConfGetStr ( " " , & defaultFS ) ; std :: string defaultScheme , defaultCluster , defaultPath ; ParseHadoopPath ( defaultFS , & defaultScheme , & defaultCluster , & defaultPath ) ; if ( scheme != defaultScheme || ( namenode . empty ( ) && namenode != defaultCluster ) ) { TF_SetStatus ( status , TF_UNIMPLEMENTED , " " ) ; return nullptr ; } namenode = " " ; } else if ( scheme == " " ) { std :: string path_har = path ; SplitArchiveNameAndPath ( & path_har , & namenode , status ) ; if ( TF_GetCode ( status ) != TF_OK ) return nullptr ; } else { if ( namenode . empty ( ) ) { namenode = " " ; } } cacheKey += namenode ; absl :: MutexLock l ( & hadoop_file -> connection_cache_lock ) ; if ( hadoop_file -> connection_cache . find ( cacheKey ) == hadoop_file -> connection_cache . end ( ) ) { hdfsBuilder * builder = libhdfs -> hdfsNewBuilder ( ) ; libhdfs -> hdfsBuilderSetNameNode ( builder , namenode . empty ( ) ? nullptr : namenode . c_str ( ) ) ; auto cacheFs = libhdfs -> hdfsBuilderConnect ( builder ) ; if ( cacheFs == nullptr ) { TF_SetStatusFromIOError ( status , TF_ABORTED , strerror ( errno ) ) ; return cacheFs ; } hadoop_file -> connection_cache [ cacheKey ] = cacheFs ; } auto fs = hadoop_file -> connection_cache [ cacheKey ] ; TF_SetStatus ( status , TF_OK , " " ) ; return fs ; }
We implement connection caching in Tensorflow, which can significantly
improve performance.
We implement connection caching in Tensorflow which can significantly improve performance .
// The default NameNode configuration will be used (from the XML // configuration files). See: // https://github.com/tensorflow/tensorflow/blob/v1.0.0/third_party/hadoop/hdfs.h#L259