@Test public void loadProfiles_attribute() { Configuration conf = new Configuration(); conf.set(key("root"), MockHadoopDataSource.class.getName()); conf.set(key("root", "path"), "/"); conf.set(key("root", "hello1"), "world1"); conf.set(key("root", "hello2"), "world2"); conf.set(key("root", "hello3"), "world3"); List<DirectDataSourceProfile> profiles = HadoopDataSourceUtil.loadProfiles(conf); assertThat(profiles.size(), is(1)); DirectDataSourceProfile profile = find(profiles, ""); assertThat(profile.getTargetClass(), equalTo((Object) MockHadoopDataSource.class)); assertThat(profile.getAttributes(), is(map("hello1", "world1", "hello2", "world2", "hello3", "world3"))); }
public static List<DirectDataSourceProfile> loadProfiles(Configuration conf) { if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> pathToKey = new HashMap<>(); Map<String, String> map = getConfigMap(conf); Set<String> keys = getChildKeys(map, "."); try { List<DirectDataSourceProfile> results = new ArrayList<>(); for (String key : keys) { String className = map.get(key); Map<String, String> config = createPrefixMap(map, key + "."); String path = config.remove(KEY_PATH); if (path == null) { throw new IllegalStateException(MessageFormat.format( "Missing I/O configuration: {0}", PREFIX + key + '.' + KEY_PATH)); } path = normalizePath(path); if (pathToKey.containsKey(path)) { throw new IllegalStateException(MessageFormat.format( "Path mapping is duplicated: {0} ({1} <=> {2})", path.isEmpty() ? "/" : path, PREFIX + key + '.' + KEY_PATH, PREFIX + pathToKey.get(key) + '.' + KEY_PATH)); } else { pathToKey.put(path, key); } Class<? extends AbstractDirectDataSource> aClass = conf.getClassByName(className) .asSubclass(AbstractDirectDataSource.class); results.add(new DirectDataSourceProfile(key, aClass, path, config)); } return results; } catch (ClassNotFoundException e) { throw new IllegalStateException(e); } }
HadoopDataSourceUtil { public static List<DirectDataSourceProfile> loadProfiles(Configuration conf) { if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> pathToKey = new HashMap<>(); Map<String, String> map = getConfigMap(conf); Set<String> keys = getChildKeys(map, "."); try { List<DirectDataSourceProfile> results = new ArrayList<>(); for (String key : keys) { String className = map.get(key); Map<String, String> config = createPrefixMap(map, key + "."); String path = config.remove(KEY_PATH); if (path == null) { throw new IllegalStateException(MessageFormat.format( "Missing I/O configuration: {0}", PREFIX + key + '.' + KEY_PATH)); } path = normalizePath(path); if (pathToKey.containsKey(path)) { throw new IllegalStateException(MessageFormat.format( "Path mapping is duplicated: {0} ({1} <=> {2})", path.isEmpty() ? "/" : path, PREFIX + key + '.' + KEY_PATH, PREFIX + pathToKey.get(key) + '.' + KEY_PATH)); } else { pathToKey.put(path, key); } Class<? extends AbstractDirectDataSource> aClass = conf.getClassByName(className) .asSubclass(AbstractDirectDataSource.class); results.add(new DirectDataSourceProfile(key, aClass, path, config)); } return results; } catch (ClassNotFoundException e) { throw new IllegalStateException(e); } } }
HadoopDataSourceUtil { public static List<DirectDataSourceProfile> loadProfiles(Configuration conf) { if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> pathToKey = new HashMap<>(); Map<String, String> map = getConfigMap(conf); Set<String> keys = getChildKeys(map, "."); try { List<DirectDataSourceProfile> results = new ArrayList<>(); for (String key : keys) { String className = map.get(key); Map<String, String> config = createPrefixMap(map, key + "."); String path = config.remove(KEY_PATH); if (path == null) { throw new IllegalStateException(MessageFormat.format( "Missing I/O configuration: {0}", PREFIX + key + '.' + KEY_PATH)); } path = normalizePath(path); if (pathToKey.containsKey(path)) { throw new IllegalStateException(MessageFormat.format( "Path mapping is duplicated: {0} ({1} <=> {2})", path.isEmpty() ? "/" : path, PREFIX + key + '.' + KEY_PATH, PREFIX + pathToKey.get(key) + '.' + KEY_PATH)); } else { pathToKey.put(path, key); } Class<? extends AbstractDirectDataSource> aClass = conf.getClassByName(className) .asSubclass(AbstractDirectDataSource.class); results.add(new DirectDataSourceProfile(key, aClass, path, config)); } return results; } catch (ClassNotFoundException e) { throw new IllegalStateException(e); } } private  HadoopDataSourceUtil(); }
HadoopDataSourceUtil { public static List<DirectDataSourceProfile> loadProfiles(Configuration conf) { if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> pathToKey = new HashMap<>(); Map<String, String> map = getConfigMap(conf); Set<String> keys = getChildKeys(map, "."); try { List<DirectDataSourceProfile> results = new ArrayList<>(); for (String key : keys) { String className = map.get(key); Map<String, String> config = createPrefixMap(map, key + "."); String path = config.remove(KEY_PATH); if (path == null) { throw new IllegalStateException(MessageFormat.format( "Missing I/O configuration: {0}", PREFIX + key + '.' + KEY_PATH)); } path = normalizePath(path); if (pathToKey.containsKey(path)) { throw new IllegalStateException(MessageFormat.format( "Path mapping is duplicated: {0} ({1} <=> {2})", path.isEmpty() ? "/" : path, PREFIX + key + '.' + KEY_PATH, PREFIX + pathToKey.get(key) + '.' + KEY_PATH)); } else { pathToKey.put(path, key); } Class<? extends AbstractDirectDataSource> aClass = conf.getClassByName(className) .asSubclass(AbstractDirectDataSource.class); results.add(new DirectDataSourceProfile(key, aClass, path, config)); } return results; } catch (ClassNotFoundException e) { throw new IllegalStateException(e); } } private  HadoopDataSourceUtil(); static List<DirectDataSourceProfile> loadProfiles(Configuration conf); static DirectDataSourceRepository loadRepository(Configuration conf); static boolean isLocalAttemptOutputDefined(LocalFileSystem localFileSystem); static Path getLocalTemporaryDirectory(LocalFileSystem localFileSystem); static OutputTransactionContext createContext(String executionId, String datasourceId); @Deprecated static OutputTransactionContext createContext(JobContext context, String datasourceId); @Deprecated static OutputAttemptContext createContext(TaskAttemptContext context, String datasourceId); static String getTransactionInfoExecutionId(Path transactionInfoPath); static Path getTransactionInfoPath(Configuration conf, String executionId); static Path getCommitMarkPath(Configuration conf, String executionId); static Collection<FileStatus> findAllTransactionInfoFiles(Configuration conf); static Path getSystemDir(Configuration conf); static Path getSystemDir(Configuration conf, boolean resolve); static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern); static List<FileStatus> onlyMinimalCovered(List<FileStatus> statList); static boolean contains(Path parent, Path child); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to); static void moveFromLocal(
            Counter counter,
            LocalFileSystem localFs, FileSystem fs,
            Path from, Path to); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to, int threads); static HadoopFileFormat<T> toHadoopFileFormat(
            Configuration configuration, DataFormat<T> format); }
HadoopDataSourceUtil { public static List<DirectDataSourceProfile> loadProfiles(Configuration conf) { if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> pathToKey = new HashMap<>(); Map<String, String> map = getConfigMap(conf); Set<String> keys = getChildKeys(map, "."); try { List<DirectDataSourceProfile> results = new ArrayList<>(); for (String key : keys) { String className = map.get(key); Map<String, String> config = createPrefixMap(map, key + "."); String path = config.remove(KEY_PATH); if (path == null) { throw new IllegalStateException(MessageFormat.format( "Missing I/O configuration: {0}", PREFIX + key + '.' + KEY_PATH)); } path = normalizePath(path); if (pathToKey.containsKey(path)) { throw new IllegalStateException(MessageFormat.format( "Path mapping is duplicated: {0} ({1} <=> {2})", path.isEmpty() ? "/" : path, PREFIX + key + '.' + KEY_PATH, PREFIX + pathToKey.get(key) + '.' + KEY_PATH)); } else { pathToKey.put(path, key); } Class<? extends AbstractDirectDataSource> aClass = conf.getClassByName(className) .asSubclass(AbstractDirectDataSource.class); results.add(new DirectDataSourceProfile(key, aClass, path, config)); } return results; } catch (ClassNotFoundException e) { throw new IllegalStateException(e); } } private  HadoopDataSourceUtil(); static List<DirectDataSourceProfile> loadProfiles(Configuration conf); static DirectDataSourceRepository loadRepository(Configuration conf); static boolean isLocalAttemptOutputDefined(LocalFileSystem localFileSystem); static Path getLocalTemporaryDirectory(LocalFileSystem localFileSystem); static OutputTransactionContext createContext(String executionId, String datasourceId); @Deprecated static OutputTransactionContext createContext(JobContext context, String datasourceId); @Deprecated static OutputAttemptContext createContext(TaskAttemptContext context, String datasourceId); static String getTransactionInfoExecutionId(Path transactionInfoPath); static Path getTransactionInfoPath(Configuration conf, String executionId); static Path getCommitMarkPath(Configuration conf, String executionId); static Collection<FileStatus> findAllTransactionInfoFiles(Configuration conf); static Path getSystemDir(Configuration conf); static Path getSystemDir(Configuration conf, boolean resolve); static List<FileStatus> search(FileSystem fs, Path base, FilePattern pattern); static List<FileStatus> onlyMinimalCovered(List<FileStatus> statList); static boolean contains(Path parent, Path child); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to); static void moveFromLocal(
            Counter counter,
            LocalFileSystem localFs, FileSystem fs,
            Path from, Path to); static void move(
            Counter counter, FileSystem fs,
            Path from, Path to, int threads); static HadoopFileFormat<T> toHadoopFileFormat(
            Configuration configuration, DataFormat<T> format); static final String PREFIX; static final String KEY_PATH; static final String KEY_SYSTEM_DIR; static final String KEY_LOCAL_TEMPDIR; static final String PREFIX_BEGIN_MARK; static final String PREFIX_COMMIT_MARK; static final String DEFAULT_SYSTEM_DIR; static final String TRANSACTION_INFO_DIR; static final Charset COMMENT_CHARSET; }