@Test public void testGetSplits() throws Exception { Configuration conf = new Configuration(); UnshardedExportToCloudStorage export = new UnshardedExportToCloudStorage( conf, "path", ExportFileFormat.AVRO, new BigQueryHelper(null), "project-id", null, new InputFormat<LongWritable, Text>() { @Override public List<InputSplit> getSplits(JobContext jobContext) { return ImmutableList.<InputSplit>builder() .add(new FileSplit(new Path("Foo"), 0L, 1L, new String[0])) .add(new FileSplit(new Path("Bar"), 0L, 1L, new String[0])) .build(); } @Override public RecordReader<LongWritable, Text> createRecordReader( InputSplit inputSplit, TaskAttemptContext taskAttemptContext) { throw new UnsupportedOperationException("Not implemented."); } }); List<InputSplit> splits = export.getSplits(null); UnshardedInputSplit fooSplit = (UnshardedInputSplit) splits.get(0); assertThat(fooSplit.getPath().getName()).isEqualTo("Foo"); UnshardedInputSplit barSplit = (UnshardedInputSplit) splits.get(1); assertThat(barSplit.getPath().getName()).isEqualTo("Bar"); }
@Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { logger.atInfo().log("Setting FileInputFormat's inputPath to '%s'", gcsPath); configuration.set(FileInputFormat.INPUT_DIR, gcsPath); List<InputSplit> splits = new ArrayList<>(); for (InputSplit split : delegateInputFormat.getSplits(context)) { FileSplit fileSplit = (FileSplit) split; splits.add( new UnshardedInputSplit( fileSplit.getPath(), fileSplit.getStart(), fileSplit.getLength(), fileSplit.getLocations())); } return splits; }
UnshardedExportToCloudStorage extends AbstractExportToCloudStorage { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { logger.atInfo().log("Setting FileInputFormat's inputPath to '%s'", gcsPath); configuration.set(FileInputFormat.INPUT_DIR, gcsPath); List<InputSplit> splits = new ArrayList<>(); for (InputSplit split : delegateInputFormat.getSplits(context)) { FileSplit fileSplit = (FileSplit) split; splits.add( new UnshardedInputSplit( fileSplit.getPath(), fileSplit.getStart(), fileSplit.getLength(), fileSplit.getLocations())); } return splits; } }
UnshardedExportToCloudStorage extends AbstractExportToCloudStorage { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { logger.atInfo().log("Setting FileInputFormat's inputPath to '%s'", gcsPath); configuration.set(FileInputFormat.INPUT_DIR, gcsPath); List<InputSplit> splits = new ArrayList<>(); for (InputSplit split : delegateInputFormat.getSplits(context)) { FileSplit fileSplit = (FileSplit) split; splits.add( new UnshardedInputSplit( fileSplit.getPath(), fileSplit.getStart(), fileSplit.getLength(), fileSplit.getLocations())); } return splits; } UnshardedExportToCloudStorage(
      Configuration configuration,
      String gcsPath,
      ExportFileFormat fileFormat,
      BigQueryHelper bigQueryHelper,
      String projectId,
      Table tableToExport,
      @Nullable InputFormat<LongWritable, Text> delegateInputFormat); }
UnshardedExportToCloudStorage extends AbstractExportToCloudStorage { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { logger.atInfo().log("Setting FileInputFormat's inputPath to '%s'", gcsPath); configuration.set(FileInputFormat.INPUT_DIR, gcsPath); List<InputSplit> splits = new ArrayList<>(); for (InputSplit split : delegateInputFormat.getSplits(context)) { FileSplit fileSplit = (FileSplit) split; splits.add( new UnshardedInputSplit( fileSplit.getPath(), fileSplit.getStart(), fileSplit.getLength(), fileSplit.getLocations())); } return splits; } UnshardedExportToCloudStorage(
      Configuration configuration,
      String gcsPath,
      ExportFileFormat fileFormat,
      BigQueryHelper bigQueryHelper,
      String projectId,
      Table tableToExport,
      @Nullable InputFormat<LongWritable, Text> delegateInputFormat); @Override List<InputSplit> getSplits(JobContext context); @Override List<String> getExportPaths(); @Override void waitForUsableMapReduceInput(); }
UnshardedExportToCloudStorage extends AbstractExportToCloudStorage { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { logger.atInfo().log("Setting FileInputFormat's inputPath to '%s'", gcsPath); configuration.set(FileInputFormat.INPUT_DIR, gcsPath); List<InputSplit> splits = new ArrayList<>(); for (InputSplit split : delegateInputFormat.getSplits(context)) { FileSplit fileSplit = (FileSplit) split; splits.add( new UnshardedInputSplit( fileSplit.getPath(), fileSplit.getStart(), fileSplit.getLength(), fileSplit.getLocations())); } return splits; } UnshardedExportToCloudStorage(
      Configuration configuration,
      String gcsPath,
      ExportFileFormat fileFormat,
      BigQueryHelper bigQueryHelper,
      String projectId,
      Table tableToExport,
      @Nullable InputFormat<LongWritable, Text> delegateInputFormat); @Override List<InputSplit> getSplits(JobContext context); @Override List<String> getExportPaths(); @Override void waitForUsableMapReduceInput();  }