public static PCollection<Read> getReadsFromBAMFilesSharded(
      Pipeline p,
      PipelineOptions pipelineOptions,
      OfflineAuth auth,
      final List<Contig> contigs,
      ReaderOptions options,
      String bamFileListOrGlob,
      final ShardingPolicy shardingPolicy) throws IOException, URISyntaxException {
      ReadBAMTransform readBAMSTransform = new ReadBAMTransform(options);
      readBAMSTransform.setAuth(auth);

      List<String> prefixes = null;
      File f = new File(bamFileListOrGlob);
      if (f.exists() && !f.isDirectory()) {
        String fileContents = Files.toString(f, Charset.defaultCharset());
        prefixes = ImmutableSet
            .<String>builder()
            .addAll(
                Splitter.on(CharMatcher.breakingWhitespace()).omitEmptyStrings().trimResults()
                    .split(fileContents))
            .build().asList();
      } else {
        prefixes = ImmutableSet
            .<String>builder()
            .add(bamFileListOrGlob)
            .build()
            .asList();
      }


      Set<String> uris = new HashSet<>();
      GcsUtil gcsUtil = pipelineOptions.as(GcsOptions.class).getGcsUtil();
      for (String prefix : prefixes) {
        URI absoluteUri = new URI(prefix);
        URI gcsUriGlob = new URI(
            absoluteUri.getScheme(),
            absoluteUri.getAuthority(),
            absoluteUri.getPath() + "*",
            absoluteUri.getQuery(),
            absoluteUri.getFragment());
        for (GcsPath entry : gcsUtil.expand(GcsPath.fromUri(gcsUriGlob))) {
          // Even if provided with an exact match to a particular BAM file, the glob operation will
          // still look for any files with that prefix, therefore also finding the corresponding
          // .bai file. Ensure only BAMs are added to the list.
          if (entry.toString().endsWith(BAMIO.BAM_FILE_SUFFIX)) {
            uris.add(entry.toString());
          }
        }
      }

      // Perform all sharding and reading in a distributed fashion, using the BreakFusion
      // transforms to ensure that work is auto-scalable based on the number of shards.
      return p
        .apply(Create.of(uris))
        .apply("Break BAM file fusion", new BreakFusionTransform<String>())
        .apply("BamsToShards", ParDo.of(new DoFn<String, BAMShard>() {
          Storage.Objects storage;
          @StartBundle
          public void startBundle(DoFn<String, BAMShard>.StartBundleContext c) throws IOException {
            storage = Transport.newStorageClient(c.getPipelineOptions().as(GCSOptions.class)).build().objects();
          }
          @DoFn.ProcessElement
          public void processElement(DoFn<String, BAMShard>.ProcessContext c) {
            List<BAMShard> shardsList = null;
            try {
              shardsList = Sharder.shardBAMFile(storage, c.element(), contigs, shardingPolicy);
              LOG.info("Sharding BAM " + c.element());
              Metrics.counter(ReadBAMTransform.class, "BAM files").inc();
              Metrics.counter(ReadBAMTransform.class, "BAM file shards").inc(shardsList.size());
            } catch (IOException e) {
              throw new RuntimeException(e);
            }
            for (BAMShard shard : shardsList) {
              c.output(shard);
            }
          }
        }))
        // We need a BreakFusionTransform here but BAMShard does not have a deterministic coder, a
        // requirement for values that are keys.  Send it as a value instead.
        .apply("Break BAMShard fusion - group", ParDo.of(new DoFn<BAMShard, KV<String, BAMShard>>() {
          @DoFn.ProcessElement
          public void processElement(DoFn<BAMShard, KV<String, BAMShard>>.ProcessContext c) throws Exception {
            c.output(KV.of(c.element().toString(), c.element()));
          }
        }))
        .apply("Break BAMShard fusion - shuffle", GroupByKey.<String, BAMShard>create())
        .apply("Break BAMShard fusion - ungroup", ParDo.of(new DoFn<KV<String, Iterable<BAMShard>>, BAMShard>() {
          @DoFn.ProcessElement
          public void processElement(DoFn<KV<String, Iterable<BAMShard>>, BAMShard>.ProcessContext c) {
            for (BAMShard shard : c.element().getValue()) {
              c.output(shard);
            }
          }
        }))
        .apply(readBAMSTransform);
  }
public static PCollection < Read > getReadsFromBAMFilesSharded ( Pipeline p , PipelineOptions pipelineOptions , OfflineAuth auth , final List < Contig > contigs , ReaderOptions options , String bamFileListOrGlob , final ShardingPolicy shardingPolicy ) throws IOException , URISyntaxException { ReadBAMTransform readBAMSTransform = new ReadBAMTransform ( options ) ; readBAMSTransform . setAuth ( auth ) ; List < String > prefixes = null ; File f = new File ( bamFileListOrGlob ) ; if ( f . exists ( ) && ! f . isDirectory ( ) ) { String fileContents = Files . toString ( f , Charset . defaultCharset ( ) ) ; prefixes = ImmutableSet . < String > builder ( ) . addAll ( Splitter . on ( CharMatcher . breakingWhitespace ( ) ) . omitEmptyStrings ( ) . trimResults ( ) . split ( fileContents ) ) . build ( ) . asList ( ) ; } else { prefixes = ImmutableSet . < String > builder ( ) . add ( bamFileListOrGlob ) . build ( ) . asList ( ) ; } Set < String > uris = new HashSet < > ( ) ; GcsUtil gcsUtil = pipelineOptions . as ( GcsOptions . class ) . getGcsUtil ( ) ; for ( String prefix : prefixes ) { URI absoluteUri = new URI ( prefix ) ; URI gcsUriGlob = new URI ( absoluteUri . getScheme ( ) , absoluteUri . getAuthority ( ) , absoluteUri . getPath ( ) + "*" , absoluteUri . getQuery ( ) , absoluteUri . getFragment ( ) ) ; for ( GcsPath entry : gcsUtil . expand ( GcsPath . fromUri ( gcsUriGlob ) ) ) { if ( entry . toString ( ) . endsWith ( BAMIO . BAM_FILE_SUFFIX ) ) { uris . add ( entry . toString ( ) ) ; } } } return p . apply ( Create . of ( uris ) ) . apply ( "Break BAM file fusion" , new BreakFusionTransform < String > ( ) ) . apply ( "BamsToShards" , ParDo . of ( new DoFn < String , BAMShard > ( ) { Storage . Objects storage ; @ StartBundle public void startBundle ( DoFn < String , BAMShard > . StartBundleContext c ) throws IOException { storage = Transport . newStorageClient ( c . getPipelineOptions ( ) . as ( GCSOptions . class ) ) . build ( ) . objects ( ) ; } @ DoFn . ProcessElement public void processElement ( DoFn < String , BAMShard > . ProcessContext c ) { List < BAMShard > shardsList = null ; try { shardsList = Sharder . shardBAMFile ( storage , c . element ( ) , contigs , shardingPolicy ) ; LOG . info ( "Sharding BAM " + c . element ( ) ) ; Metrics . counter ( ReadBAMTransform . class , "BAM files" ) . inc ( ) ; Metrics . counter ( ReadBAMTransform . class , "BAM file shards" ) . inc ( shardsList . size ( ) ) ; } catch ( IOException e ) { throw new RuntimeException ( e ) ; } for ( BAMShard shard : shardsList ) { c . output ( shard ) ; } } } ) ) . apply ( "Break BAMShard fusion - group" , ParDo . of ( new DoFn < BAMShard , KV < String , BAMShard > > ( ) { @ DoFn . ProcessElement public void processElement ( DoFn < BAMShard , KV < String , BAMShard > > . ProcessContext c ) throws Exception { c . output ( KV . of ( c . element ( ) . toString ( ) , c . element ( ) ) ) ; } } ) ) . apply ( "Break BAMShard fusion - shuffle" , GroupByKey . < String , BAMShard > create ( ) ) . apply ( "Break BAMShard fusion - ungroup" , ParDo . of ( new DoFn < KV < String , Iterable < BAMShard > > , BAMShard > ( ) { @ DoFn . ProcessElement public void processElement ( DoFn < KV < String , Iterable < BAMShard > > , BAMShard > . ProcessContext c ) { for ( BAMShard shard : c . element ( ) . getValue ( ) ) { c . output ( shard ) ; } } } ) ) . apply ( readBAMSTransform ) ; }
Get reads from one or more BAM files by reading shards in parallel.

@param p
@param pipelineOptions
@param auth
@param contigs
@param options
@param bamFileListOrGlob
@param shardingPolicy
@return
@throws IOException
@throws URISyntaxException