@Test public void input_large() throws Exception { long fragmentSize = 1 * 1024 * 1024; int fragmentCount = 20; put(new File(mapping, "input/file.txt"), fragmentSize * fragmentCount); profile.setMinimumFragmentSize(1); profile.setPreferredFragmentSize(fragmentSize); HadoopDataSourceCore core = new HadoopDataSourceCore(profile); List<DirectInputFragment> fragments = core.findInputFragments( definition, "input", FilePattern.compile("**")); assertThat(fragments.size(), is(greaterThanOrEqualTo(fragmentCount / 2))); for (DirectInputFragment fragment : fragments) { assertThat(fragment.getSize(), is(greaterThanOrEqualTo(fragmentSize / 2))); assertThat(fragment.getSize(), is(lessThanOrEqualTo(fragmentSize * 2))); } }
@Override public <T> List<DirectInputFragment> findInputFragments( DataDefinition<T> definition, String basePath, ResourcePattern resourcePattern) throws IOException, InterruptedException { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start finding input (id={0}, path={1}, resourcePattern={2})", profile.getId(), basePath, resourcePattern)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); Path temporary = p.getTemporaryFileSystemPath(); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); stats = filesOnly(stats, temporary); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process finding input (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } if (LOG.isTraceEnabled()) { for (FileStatus stat : stats) { LOG.trace(MessageFormat.format( "Input found (path={0}, length={1})", stat.getPath(), stat.getLen())); } } DataFilter<?> filter = definition.getDataFilter(); if (filter != null) { stats = applyFilter(stats, filter); } DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); List<DirectInputFragment> results; if (format instanceof StripedDataFormat<?>) { StripedDataFormat.InputContext context = new StripedDataFormat.InputContext( dataType, stats, fs, p.getMinimumFragmentSize(), p.getPreferredFragmentSize(), p.isSplitBlocks(), p.isCombineBlocks()); StripedDataFormat<T> sformat = (StripedDataFormat<T>) format; results = sformat.computeInputFragments(context); } else if (format instanceof FragmentableDataFormat<?>) { FragmentableDataFormat<T> sformat = (FragmentableDataFormat<T>) format; FragmentComputer optimizer = new FragmentComputer( p.getMinimumFragmentSize(sformat), p.getPreferredFragmentSize(sformat), p.isCombineBlocks(), p.isSplitBlocks()); results = computeInputFragments(optimizer, stats); } else { FragmentComputer optimizer = new FragmentComputer(); results = computeInputFragments(optimizer, stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish finding input (id={0}, path={1}, resource={2}, fragments={3})", profile.getId(), basePath, resourcePattern, results.size())); } return results; }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> List<DirectInputFragment> findInputFragments( DataDefinition<T> definition, String basePath, ResourcePattern resourcePattern) throws IOException, InterruptedException { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start finding input (id={0}, path={1}, resourcePattern={2})", profile.getId(), basePath, resourcePattern)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); Path temporary = p.getTemporaryFileSystemPath(); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); stats = filesOnly(stats, temporary); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process finding input (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } if (LOG.isTraceEnabled()) { for (FileStatus stat : stats) { LOG.trace(MessageFormat.format( "Input found (path={0}, length={1})", stat.getPath(), stat.getLen())); } } DataFilter<?> filter = definition.getDataFilter(); if (filter != null) { stats = applyFilter(stats, filter); } DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); List<DirectInputFragment> results; if (format instanceof StripedDataFormat<?>) { StripedDataFormat.InputContext context = new StripedDataFormat.InputContext( dataType, stats, fs, p.getMinimumFragmentSize(), p.getPreferredFragmentSize(), p.isSplitBlocks(), p.isCombineBlocks()); StripedDataFormat<T> sformat = (StripedDataFormat<T>) format; results = sformat.computeInputFragments(context); } else if (format instanceof FragmentableDataFormat<?>) { FragmentableDataFormat<T> sformat = (FragmentableDataFormat<T>) format; FragmentComputer optimizer = new FragmentComputer( p.getMinimumFragmentSize(sformat), p.getPreferredFragmentSize(sformat), p.isCombineBlocks(), p.isSplitBlocks()); results = computeInputFragments(optimizer, stats); } else { FragmentComputer optimizer = new FragmentComputer(); results = computeInputFragments(optimizer, stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish finding input (id={0}, path={1}, resource={2}, fragments={3})", profile.getId(), basePath, resourcePattern, results.size())); } return results; } }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> List<DirectInputFragment> findInputFragments( DataDefinition<T> definition, String basePath, ResourcePattern resourcePattern) throws IOException, InterruptedException { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start finding input (id={0}, path={1}, resourcePattern={2})", profile.getId(), basePath, resourcePattern)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); Path temporary = p.getTemporaryFileSystemPath(); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); stats = filesOnly(stats, temporary); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process finding input (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } if (LOG.isTraceEnabled()) { for (FileStatus stat : stats) { LOG.trace(MessageFormat.format( "Input found (path={0}, length={1})", stat.getPath(), stat.getLen())); } } DataFilter<?> filter = definition.getDataFilter(); if (filter != null) { stats = applyFilter(stats, filter); } DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); List<DirectInputFragment> results; if (format instanceof StripedDataFormat<?>) { StripedDataFormat.InputContext context = new StripedDataFormat.InputContext( dataType, stats, fs, p.getMinimumFragmentSize(), p.getPreferredFragmentSize(), p.isSplitBlocks(), p.isCombineBlocks()); StripedDataFormat<T> sformat = (StripedDataFormat<T>) format; results = sformat.computeInputFragments(context); } else if (format instanceof FragmentableDataFormat<?>) { FragmentableDataFormat<T> sformat = (FragmentableDataFormat<T>) format; FragmentComputer optimizer = new FragmentComputer( p.getMinimumFragmentSize(sformat), p.getPreferredFragmentSize(sformat), p.isCombineBlocks(), p.isSplitBlocks()); results = computeInputFragments(optimizer, stats); } else { FragmentComputer optimizer = new FragmentComputer(); results = computeInputFragments(optimizer, stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish finding input (id={0}, path={1}, resource={2}, fragments={3})", profile.getId(), basePath, resourcePattern, results.size())); } return results; } HadoopDataSourceCore(HadoopDataSourceProfile profile); }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> List<DirectInputFragment> findInputFragments( DataDefinition<T> definition, String basePath, ResourcePattern resourcePattern) throws IOException, InterruptedException { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start finding input (id={0}, path={1}, resourcePattern={2})", profile.getId(), basePath, resourcePattern)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); Path temporary = p.getTemporaryFileSystemPath(); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); stats = filesOnly(stats, temporary); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process finding input (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } if (LOG.isTraceEnabled()) { for (FileStatus stat : stats) { LOG.trace(MessageFormat.format( "Input found (path={0}, length={1})", stat.getPath(), stat.getLen())); } } DataFilter<?> filter = definition.getDataFilter(); if (filter != null) { stats = applyFilter(stats, filter); } DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); List<DirectInputFragment> results; if (format instanceof StripedDataFormat<?>) { StripedDataFormat.InputContext context = new StripedDataFormat.InputContext( dataType, stats, fs, p.getMinimumFragmentSize(), p.getPreferredFragmentSize(), p.isSplitBlocks(), p.isCombineBlocks()); StripedDataFormat<T> sformat = (StripedDataFormat<T>) format; results = sformat.computeInputFragments(context); } else if (format instanceof FragmentableDataFormat<?>) { FragmentableDataFormat<T> sformat = (FragmentableDataFormat<T>) format; FragmentComputer optimizer = new FragmentComputer( p.getMinimumFragmentSize(sformat), p.getPreferredFragmentSize(sformat), p.isCombineBlocks(), p.isSplitBlocks()); results = computeInputFragments(optimizer, stats); } else { FragmentComputer optimizer = new FragmentComputer(); results = computeInputFragments(optimizer, stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish finding input (id={0}, path={1}, resource={2}, fragments={3})", profile.getId(), basePath, resourcePattern, results.size())); } return results; } HadoopDataSourceCore(HadoopDataSourceProfile profile); @Override String path(String basePath, ResourcePattern resourcePattern); @Override String path(String basePath); @Override List<DirectInputFragment> findInputFragments(
            DataDefinition<T> definition,
            String basePath,
            ResourcePattern resourcePattern); @Override ModelInput<T> openInput(
            DataDefinition<T> definition,
            DirectInputFragment fragment,
            Counter counter); @Override ModelOutput<T> openOutput(
            OutputAttemptContext context,
            DataDefinition<T> definition,
            String basePath,
            String resourcePath,
            Counter counter); @Override List<ResourceInfo> list(
            String basePath,
            ResourcePattern resourcePattern,
            Counter counter); @Override boolean delete(
            String basePath,
            ResourcePattern resourcePattern,
            boolean recursive,
            Counter counter); @Override void setupAttemptOutput(OutputAttemptContext context); @Override void commitAttemptOutput(OutputAttemptContext context); @Override void cleanupAttemptOutput(OutputAttemptContext context); @Override void setupTransactionOutput(OutputTransactionContext context); @Override void commitTransactionOutput(OutputTransactionContext context); @Override void cleanupTransactionOutput(OutputTransactionContext context); @Override Optional<T> findProperty(Class<T> propertyType); }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> List<DirectInputFragment> findInputFragments( DataDefinition<T> definition, String basePath, ResourcePattern resourcePattern) throws IOException, InterruptedException { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start finding input (id={0}, path={1}, resourcePattern={2})", profile.getId(), basePath, resourcePattern)); } FilePattern pattern = validate(resourcePattern); HadoopDataSourceProfile p = profile; FileSystem fs = p.getFileSystem(); Path root = p.getFileSystemPath(); Path base = append(root, basePath); Path temporary = p.getTemporaryFileSystemPath(); List<FileStatus> stats = HadoopDataSourceUtil.search(fs, base, pattern); stats = filesOnly(stats, temporary); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Process finding input (id={0}, path={1}, resource={2}, files={3})", profile.getId(), basePath, resourcePattern, stats.size())); } if (LOG.isTraceEnabled()) { for (FileStatus stat : stats) { LOG.trace(MessageFormat.format( "Input found (path={0}, length={1})", stat.getPath(), stat.getLen())); } } DataFilter<?> filter = definition.getDataFilter(); if (filter != null) { stats = applyFilter(stats, filter); } DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); List<DirectInputFragment> results; if (format instanceof StripedDataFormat<?>) { StripedDataFormat.InputContext context = new StripedDataFormat.InputContext( dataType, stats, fs, p.getMinimumFragmentSize(), p.getPreferredFragmentSize(), p.isSplitBlocks(), p.isCombineBlocks()); StripedDataFormat<T> sformat = (StripedDataFormat<T>) format; results = sformat.computeInputFragments(context); } else if (format instanceof FragmentableDataFormat<?>) { FragmentableDataFormat<T> sformat = (FragmentableDataFormat<T>) format; FragmentComputer optimizer = new FragmentComputer( p.getMinimumFragmentSize(sformat), p.getPreferredFragmentSize(sformat), p.isCombineBlocks(), p.isSplitBlocks()); results = computeInputFragments(optimizer, stats); } else { FragmentComputer optimizer = new FragmentComputer(); results = computeInputFragments(optimizer, stats); } if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish finding input (id={0}, path={1}, resource={2}, fragments={3})", profile.getId(), basePath, resourcePattern, results.size())); } return results; } HadoopDataSourceCore(HadoopDataSourceProfile profile); @Override String path(String basePath, ResourcePattern resourcePattern); @Override String path(String basePath); @Override List<DirectInputFragment> findInputFragments(
            DataDefinition<T> definition,
            String basePath,
            ResourcePattern resourcePattern); @Override ModelInput<T> openInput(
            DataDefinition<T> definition,
            DirectInputFragment fragment,
            Counter counter); @Override ModelOutput<T> openOutput(
            OutputAttemptContext context,
            DataDefinition<T> definition,
            String basePath,
            String resourcePath,
            Counter counter); @Override List<ResourceInfo> list(
            String basePath,
            ResourcePattern resourcePattern,
            Counter counter); @Override boolean delete(
            String basePath,
            ResourcePattern resourcePattern,
            boolean recursive,
            Counter counter); @Override void setupAttemptOutput(OutputAttemptContext context); @Override void commitAttemptOutput(OutputAttemptContext context); @Override void cleanupAttemptOutput(OutputAttemptContext context); @Override void setupTransactionOutput(OutputTransactionContext context); @Override void commitTransactionOutput(OutputTransactionContext context); @Override void cleanupTransactionOutput(OutputTransactionContext context); @Override Optional<T> findProperty(Class<T> propertyType);  }