@Test public void output_multirecord() throws Exception { HadoopDataSourceCore core = new HadoopDataSourceCore(profile); setup(core); try (ModelOutput<StringBuilder> output = core.openOutput( context, definition, "output", "file.txt", counter)) { output.write(new StringBuilder("Hello, world!")); } File target = new File(mapping, "output/file.txt"); assertThat(target.exists(), is(false)); commitAttempt(core); assertThat(target.exists(), is(false)); commitTransaction(core); assertThat(target.exists(), is(true)); assertThat(get(target), is(Arrays.asList("Hello, world!"))); }
@Override public <T> ModelOutput<T> openOutput( OutputAttemptContext context, DataDefinition<T> definition, String basePath, String resourcePath, Counter counter) throws IOException, InterruptedException { boolean local = isLocalAttemptOutput(); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start opening output (id={0}, path={1}, resource={2}, streaming={3})", profile.getId(), basePath, resourcePath, local == false)); } FileSystem fs = local ? profile.getLocalFileSystem() : profile.getFileSystem(); Path attempt = local ? getLocalAttemptOutput(context) : getAttemptOutput(context); DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); Path file = append(append(attempt, basePath), resourcePath); HadoopFileFormat<T> fileFormat = convertFormat(format); ModelOutput<T> output = fileFormat.createOutput(dataType, fs, file, counter); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish opening output (id={0}, path={1}, resource={2}, file={3})", profile.getId(), basePath, resourcePath, file)); } return output; }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> ModelOutput<T> openOutput( OutputAttemptContext context, DataDefinition<T> definition, String basePath, String resourcePath, Counter counter) throws IOException, InterruptedException { boolean local = isLocalAttemptOutput(); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start opening output (id={0}, path={1}, resource={2}, streaming={3})", profile.getId(), basePath, resourcePath, local == false)); } FileSystem fs = local ? profile.getLocalFileSystem() : profile.getFileSystem(); Path attempt = local ? getLocalAttemptOutput(context) : getAttemptOutput(context); DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); Path file = append(append(attempt, basePath), resourcePath); HadoopFileFormat<T> fileFormat = convertFormat(format); ModelOutput<T> output = fileFormat.createOutput(dataType, fs, file, counter); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish opening output (id={0}, path={1}, resource={2}, file={3})", profile.getId(), basePath, resourcePath, file)); } return output; } }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> ModelOutput<T> openOutput( OutputAttemptContext context, DataDefinition<T> definition, String basePath, String resourcePath, Counter counter) throws IOException, InterruptedException { boolean local = isLocalAttemptOutput(); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start opening output (id={0}, path={1}, resource={2}, streaming={3})", profile.getId(), basePath, resourcePath, local == false)); } FileSystem fs = local ? profile.getLocalFileSystem() : profile.getFileSystem(); Path attempt = local ? getLocalAttemptOutput(context) : getAttemptOutput(context); DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); Path file = append(append(attempt, basePath), resourcePath); HadoopFileFormat<T> fileFormat = convertFormat(format); ModelOutput<T> output = fileFormat.createOutput(dataType, fs, file, counter); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish opening output (id={0}, path={1}, resource={2}, file={3})", profile.getId(), basePath, resourcePath, file)); } return output; } HadoopDataSourceCore(HadoopDataSourceProfile profile); }
HadoopDataSourceCore implements DirectDataSource { @Override public <T> ModelOutput<T> openOutput( OutputAttemptContext context, DataDefinition<T> definition, String basePath, String resourcePath, Counter counter) throws IOException, InterruptedException { boolean local = isLocalAttemptOutput(); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start opening output (id={0}, path={1}, resource={2}, streaming={3})", profile.getId(), basePath, resourcePath, local == false)); } FileSystem fs = local ? profile.getLocalFileSystem() : profile.getFileSystem(); Path attempt = local ? getLocalAttemptOutput(context) : getAttemptOutput(context); DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); Path file = append(append(attempt, basePath), resourcePath); HadoopFileFormat<T> fileFormat = convertFormat(format); ModelOutput<T> output = fileFormat.createOutput(dataType, fs, file, counter); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish opening output (id={0}, path={1}, resource={2}, file={3})", profile.getId(), basePath, resourcePath, file)); } return output; } 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> ModelOutput<T> openOutput( OutputAttemptContext context, DataDefinition<T> definition, String basePath, String resourcePath, Counter counter) throws IOException, InterruptedException { boolean local = isLocalAttemptOutput(); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Start opening output (id={0}, path={1}, resource={2}, streaming={3})", profile.getId(), basePath, resourcePath, local == false)); } FileSystem fs = local ? profile.getLocalFileSystem() : profile.getFileSystem(); Path attempt = local ? getLocalAttemptOutput(context) : getAttemptOutput(context); DataFormat<T> format = definition.getDataFormat(); Class<? extends T> dataType = definition.getDataClass(); Path file = append(append(attempt, basePath), resourcePath); HadoopFileFormat<T> fileFormat = convertFormat(format); ModelOutput<T> output = fileFormat.createOutput(dataType, fs, file, counter); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Finish opening output (id={0}, path={1}, resource={2}, file={3})", profile.getId(), basePath, resourcePath, file)); } return output; } 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);  }