@Test public void testBasicImport() throws Exception { command.targets = Lists.newArrayList(sample, datasetName); command.run(); Assert.assertEquals("Should contain expected records", expected, DatasetTestUtilities.materialize(dataset)); verify(console).trace(contains("repo:file:target/data")); verify(console).info("Added {} records to \"{}\"", 2l, datasetName); verifyNoMoreInteractions(console); }
@Override @SuppressWarnings("unchecked") public int run() throws IOException { Preconditions.checkArgument(targets != null && targets.size() == 2, "CSV path and target dataset name are required."); Path source = qualifiedPath(targets.get(0)); FileSystem sourceFS = source.getFileSystem(getConf()); Preconditions.checkArgument(sourceFS.exists(source), "CSV path does not exist: " + source); if (header != null) { noHeader = true; } CSVProperties props = new CSVProperties.Builder() .delimiter(delimiter) .escape(escape) .quote(quote) .header(header) .hasHeader(!noHeader) .linesToSkip(linesToSkip) .charset(charsetName) .build(); String dataset = targets.get(1); View<Record> target = load(dataset, Record.class); Schema datasetSchema = target.getDataset().getDescriptor().getSchema(); DatasetDescriptor csvDescriptor = new DatasetDescriptor.Builder() .location(source.toUri()) .schema(ColumnMappingParser.removeEmbeddedMapping( PartitionStrategyParser.removeEmbeddedStrategy(datasetSchema))) .format("csv") .build(); csvDescriptor = props.addToDescriptor(csvDescriptor); TemporaryFileSystemDatasetRepository repo = new TemporaryFileSystemDatasetRepository(getConf(), sourceFS.makeQualified(new Path("/tmp/" + UUID.randomUUID().toString())), target.getDataset().getNamespace(), UUID.randomUUID().toString()); try { FileSystemDataset<Record> csvDataset = (FileSystemDataset) repo.create("default", "csv", csvDescriptor); Iterator<Path> iter = csvDataset.pathIterator().iterator(); Preconditions.checkArgument(iter.hasNext(), "CSV path has no data files: " + source); TaskUtil.configure(getConf()).addJars(jars); TransformTask task; if (transform != null) { DoFn<Record, Record> transformFn; try { DynConstructors.Ctor<DoFn<Record, Record>> ctor = new DynConstructors.Builder(DoFn.class) .loader(loaderForJars(jars)) .impl(transform) .buildChecked(); transformFn = ctor.newInstance(); } catch (NoSuchMethodException e) { throw new DatasetException( "Cannot find no-arg constructor for class: " + transform, e); } task = new TransformTask<Record, Record>( csvDataset, target, transformFn); } else { task = new CopyTask<Record>(csvDataset, target); } task.setConf(getConf()); if (noCompaction) { task.noCompaction(); } if (numWriters >= 0) { task.setNumWriters(numWriters); } if (filesPerPartition > 0) { task.setFilesPerPartition(filesPerPartition); } if (overwrite) { task.setWriteMode(Target.WriteMode.OVERWRITE); } PipelineResult result = task.run(); if (result.succeeded()) { long count = task.getCount(); if (count > 0) { console.info("Added {} records to \"{}\"", count, dataset); } return 0; } else { return 1; } } finally { repo.delete(); } }
CSVImportCommand extends BaseDatasetCommand { @Override @SuppressWarnings("unchecked") public int run() throws IOException { Preconditions.checkArgument(targets != null && targets.size() == 2, "CSV path and target dataset name are required."); Path source = qualifiedPath(targets.get(0)); FileSystem sourceFS = source.getFileSystem(getConf()); Preconditions.checkArgument(sourceFS.exists(source), "CSV path does not exist: " + source); if (header != null) { noHeader = true; } CSVProperties props = new CSVProperties.Builder() .delimiter(delimiter) .escape(escape) .quote(quote) .header(header) .hasHeader(!noHeader) .linesToSkip(linesToSkip) .charset(charsetName) .build(); String dataset = targets.get(1); View<Record> target = load(dataset, Record.class); Schema datasetSchema = target.getDataset().getDescriptor().getSchema(); DatasetDescriptor csvDescriptor = new DatasetDescriptor.Builder() .location(source.toUri()) .schema(ColumnMappingParser.removeEmbeddedMapping( PartitionStrategyParser.removeEmbeddedStrategy(datasetSchema))) .format("csv") .build(); csvDescriptor = props.addToDescriptor(csvDescriptor); TemporaryFileSystemDatasetRepository repo = new TemporaryFileSystemDatasetRepository(getConf(), sourceFS.makeQualified(new Path("/tmp/" + UUID.randomUUID().toString())), target.getDataset().getNamespace(), UUID.randomUUID().toString()); try { FileSystemDataset<Record> csvDataset = (FileSystemDataset) repo.create("default", "csv", csvDescriptor); Iterator<Path> iter = csvDataset.pathIterator().iterator(); Preconditions.checkArgument(iter.hasNext(), "CSV path has no data files: " + source); TaskUtil.configure(getConf()).addJars(jars); TransformTask task; if (transform != null) { DoFn<Record, Record> transformFn; try { DynConstructors.Ctor<DoFn<Record, Record>> ctor = new DynConstructors.Builder(DoFn.class) .loader(loaderForJars(jars)) .impl(transform) .buildChecked(); transformFn = ctor.newInstance(); } catch (NoSuchMethodException e) { throw new DatasetException( "Cannot find no-arg constructor for class: " + transform, e); } task = new TransformTask<Record, Record>( csvDataset, target, transformFn); } else { task = new CopyTask<Record>(csvDataset, target); } task.setConf(getConf()); if (noCompaction) { task.noCompaction(); } if (numWriters >= 0) { task.setNumWriters(numWriters); } if (filesPerPartition > 0) { task.setFilesPerPartition(filesPerPartition); } if (overwrite) { task.setWriteMode(Target.WriteMode.OVERWRITE); } PipelineResult result = task.run(); if (result.succeeded()) { long count = task.getCount(); if (count > 0) { console.info("Added {} records to \"{}\"", count, dataset); } return 0; } else { return 1; } } finally { repo.delete(); } } }
CSVImportCommand extends BaseDatasetCommand { @Override @SuppressWarnings("unchecked") public int run() throws IOException { Preconditions.checkArgument(targets != null && targets.size() == 2, "CSV path and target dataset name are required."); Path source = qualifiedPath(targets.get(0)); FileSystem sourceFS = source.getFileSystem(getConf()); Preconditions.checkArgument(sourceFS.exists(source), "CSV path does not exist: " + source); if (header != null) { noHeader = true; } CSVProperties props = new CSVProperties.Builder() .delimiter(delimiter) .escape(escape) .quote(quote) .header(header) .hasHeader(!noHeader) .linesToSkip(linesToSkip) .charset(charsetName) .build(); String dataset = targets.get(1); View<Record> target = load(dataset, Record.class); Schema datasetSchema = target.getDataset().getDescriptor().getSchema(); DatasetDescriptor csvDescriptor = new DatasetDescriptor.Builder() .location(source.toUri()) .schema(ColumnMappingParser.removeEmbeddedMapping( PartitionStrategyParser.removeEmbeddedStrategy(datasetSchema))) .format("csv") .build(); csvDescriptor = props.addToDescriptor(csvDescriptor); TemporaryFileSystemDatasetRepository repo = new TemporaryFileSystemDatasetRepository(getConf(), sourceFS.makeQualified(new Path("/tmp/" + UUID.randomUUID().toString())), target.getDataset().getNamespace(), UUID.randomUUID().toString()); try { FileSystemDataset<Record> csvDataset = (FileSystemDataset) repo.create("default", "csv", csvDescriptor); Iterator<Path> iter = csvDataset.pathIterator().iterator(); Preconditions.checkArgument(iter.hasNext(), "CSV path has no data files: " + source); TaskUtil.configure(getConf()).addJars(jars); TransformTask task; if (transform != null) { DoFn<Record, Record> transformFn; try { DynConstructors.Ctor<DoFn<Record, Record>> ctor = new DynConstructors.Builder(DoFn.class) .loader(loaderForJars(jars)) .impl(transform) .buildChecked(); transformFn = ctor.newInstance(); } catch (NoSuchMethodException e) { throw new DatasetException( "Cannot find no-arg constructor for class: " + transform, e); } task = new TransformTask<Record, Record>( csvDataset, target, transformFn); } else { task = new CopyTask<Record>(csvDataset, target); } task.setConf(getConf()); if (noCompaction) { task.noCompaction(); } if (numWriters >= 0) { task.setNumWriters(numWriters); } if (filesPerPartition > 0) { task.setFilesPerPartition(filesPerPartition); } if (overwrite) { task.setWriteMode(Target.WriteMode.OVERWRITE); } PipelineResult result = task.run(); if (result.succeeded()) { long count = task.getCount(); if (count > 0) { console.info("Added {} records to \"{}\"", count, dataset); } return 0; } else { return 1; } } finally { repo.delete(); } } CSVImportCommand(Logger console); }
CSVImportCommand extends BaseDatasetCommand { @Override @SuppressWarnings("unchecked") public int run() throws IOException { Preconditions.checkArgument(targets != null && targets.size() == 2, "CSV path and target dataset name are required."); Path source = qualifiedPath(targets.get(0)); FileSystem sourceFS = source.getFileSystem(getConf()); Preconditions.checkArgument(sourceFS.exists(source), "CSV path does not exist: " + source); if (header != null) { noHeader = true; } CSVProperties props = new CSVProperties.Builder() .delimiter(delimiter) .escape(escape) .quote(quote) .header(header) .hasHeader(!noHeader) .linesToSkip(linesToSkip) .charset(charsetName) .build(); String dataset = targets.get(1); View<Record> target = load(dataset, Record.class); Schema datasetSchema = target.getDataset().getDescriptor().getSchema(); DatasetDescriptor csvDescriptor = new DatasetDescriptor.Builder() .location(source.toUri()) .schema(ColumnMappingParser.removeEmbeddedMapping( PartitionStrategyParser.removeEmbeddedStrategy(datasetSchema))) .format("csv") .build(); csvDescriptor = props.addToDescriptor(csvDescriptor); TemporaryFileSystemDatasetRepository repo = new TemporaryFileSystemDatasetRepository(getConf(), sourceFS.makeQualified(new Path("/tmp/" + UUID.randomUUID().toString())), target.getDataset().getNamespace(), UUID.randomUUID().toString()); try { FileSystemDataset<Record> csvDataset = (FileSystemDataset) repo.create("default", "csv", csvDescriptor); Iterator<Path> iter = csvDataset.pathIterator().iterator(); Preconditions.checkArgument(iter.hasNext(), "CSV path has no data files: " + source); TaskUtil.configure(getConf()).addJars(jars); TransformTask task; if (transform != null) { DoFn<Record, Record> transformFn; try { DynConstructors.Ctor<DoFn<Record, Record>> ctor = new DynConstructors.Builder(DoFn.class) .loader(loaderForJars(jars)) .impl(transform) .buildChecked(); transformFn = ctor.newInstance(); } catch (NoSuchMethodException e) { throw new DatasetException( "Cannot find no-arg constructor for class: " + transform, e); } task = new TransformTask<Record, Record>( csvDataset, target, transformFn); } else { task = new CopyTask<Record>(csvDataset, target); } task.setConf(getConf()); if (noCompaction) { task.noCompaction(); } if (numWriters >= 0) { task.setNumWriters(numWriters); } if (filesPerPartition > 0) { task.setFilesPerPartition(filesPerPartition); } if (overwrite) { task.setWriteMode(Target.WriteMode.OVERWRITE); } PipelineResult result = task.run(); if (result.succeeded()) { long count = task.getCount(); if (count > 0) { console.info("Added {} records to \"{}\"", count, dataset); } return 0; } else { return 1; } } finally { repo.delete(); } } CSVImportCommand(Logger console); @Override @SuppressWarnings("unchecked") int run(); @Override List<String> getExamples(); }
CSVImportCommand extends BaseDatasetCommand { @Override @SuppressWarnings("unchecked") public int run() throws IOException { Preconditions.checkArgument(targets != null && targets.size() == 2, "CSV path and target dataset name are required."); Path source = qualifiedPath(targets.get(0)); FileSystem sourceFS = source.getFileSystem(getConf()); Preconditions.checkArgument(sourceFS.exists(source), "CSV path does not exist: " + source); if (header != null) { noHeader = true; } CSVProperties props = new CSVProperties.Builder() .delimiter(delimiter) .escape(escape) .quote(quote) .header(header) .hasHeader(!noHeader) .linesToSkip(linesToSkip) .charset(charsetName) .build(); String dataset = targets.get(1); View<Record> target = load(dataset, Record.class); Schema datasetSchema = target.getDataset().getDescriptor().getSchema(); DatasetDescriptor csvDescriptor = new DatasetDescriptor.Builder() .location(source.toUri()) .schema(ColumnMappingParser.removeEmbeddedMapping( PartitionStrategyParser.removeEmbeddedStrategy(datasetSchema))) .format("csv") .build(); csvDescriptor = props.addToDescriptor(csvDescriptor); TemporaryFileSystemDatasetRepository repo = new TemporaryFileSystemDatasetRepository(getConf(), sourceFS.makeQualified(new Path("/tmp/" + UUID.randomUUID().toString())), target.getDataset().getNamespace(), UUID.randomUUID().toString()); try { FileSystemDataset<Record> csvDataset = (FileSystemDataset) repo.create("default", "csv", csvDescriptor); Iterator<Path> iter = csvDataset.pathIterator().iterator(); Preconditions.checkArgument(iter.hasNext(), "CSV path has no data files: " + source); TaskUtil.configure(getConf()).addJars(jars); TransformTask task; if (transform != null) { DoFn<Record, Record> transformFn; try { DynConstructors.Ctor<DoFn<Record, Record>> ctor = new DynConstructors.Builder(DoFn.class) .loader(loaderForJars(jars)) .impl(transform) .buildChecked(); transformFn = ctor.newInstance(); } catch (NoSuchMethodException e) { throw new DatasetException( "Cannot find no-arg constructor for class: " + transform, e); } task = new TransformTask<Record, Record>( csvDataset, target, transformFn); } else { task = new CopyTask<Record>(csvDataset, target); } task.setConf(getConf()); if (noCompaction) { task.noCompaction(); } if (numWriters >= 0) { task.setNumWriters(numWriters); } if (filesPerPartition > 0) { task.setFilesPerPartition(filesPerPartition); } if (overwrite) { task.setWriteMode(Target.WriteMode.OVERWRITE); } PipelineResult result = task.run(); if (result.succeeded()) { long count = task.getCount(); if (count > 0) { console.info("Added {} records to \"{}\"", count, dataset); } return 0; } else { return 1; } } finally { repo.delete(); } } CSVImportCommand(Logger console); @Override @SuppressWarnings("unchecked") int run(); @Override List<String> getExamples();  }