Merge pull request googlegenomics#47 from deflaux/master

Update to latest version of DataflowSDK.

Author: deflaux

README.rst

@@ -31,7 +31,7 @@ Getting started
   Project ID and Google Cloud Storage bucket you made in the first step.
   This command runs the VariantSimilarity pipeline (which runs PCoA on a dataset)::
 
-    java -cp target/google-genomics-dataflow-v1beta2-0.2-SNAPSHOT.jar \
+    java -cp target/google-genomics-dataflow-*.jar \
       com.google.cloud.genomics.dataflow.pipelines.VariantSimilarity \
       --project=my-project-id \
       --output=gs://my-bucket/output/localtest.txt \
@@ -43,7 +43,7 @@ Getting started
 * To deploy your pipeline (which runs on Google Compute Engine), some additional
   command line arguments are required::
 
-    java -cp target/google-genomics-dataflow-v1beta2-0.2-SNAPSHOT.jar \
+    java -cp target/google-genomics-dataflow-*.jar \
       com.google.cloud.genomics.dataflow.pipelines.VariantSimilarity \
       --runner=BlockingDataflowPipelineRunner \
       --project=my-project-id \
@@ -65,7 +65,7 @@ Identity By State (IBS)
   first argument provided in the above command lines. For example, to run Identity by State
   change ``VariantSimilarity`` to ``IdentityByState``::
 
-    java -cp target/google-genomics-dataflow-v1beta2-0.2-SNAPSHOT.jar \
+    java -cp target/google-genomics-dataflow-*.jar \
       com.google.cloud.genomics.dataflow.pipelines.IdentityByState \
       --project=my-project-id \
       --output=gs://my-bucket/localtest.txt \
@@ -122,15 +122,26 @@ To depend on this code, add the following to your ``pom.xml`` file::
       <dependency>
         <groupId>com.google.cloud.genomics</groupId>
         <artifactId>google-genomics-dataflow</artifactId>
-        <version>v1beta2-0.2</version>
+        <version>LATEST</version>
       </dependency>
     </dependencies>
   </project>
 
 You can find the latest version in
 `Maven's central repository <https://search.maven.org/#search%7Cga%7C1%7Ca%3A%22google-genomics-dataflow%22>`_
 
-We'll soon include an example pipeline that depends on this code in another GitHub repository.
+For an example pipeline that depends on this code in another GitHub repository, see https://github.com/googlegenomics/codelabs/tree/master/Java/PlatinumGenomes-variant-transformation.
+
+Updating the prebuilt jar
+-------------------------
+
+The prebuilt jar is currently generated by hand, whenever the code seems stable and has enough new features to warrant the effort.
+
+To update the jar, we use a different set of maven commands:
+
+  cd dataflow-java
+  mvn clean compile bundle:bundle
+  cp target/google-genomics-dataflow-*.jar google-genomics-dataflow.jar
 
 Project status
 --------------

pom.xml

@@ -69,7 +69,7 @@
     <dependency>
       <groupId>com.google.cloud.dataflow</groupId>
       <artifactId>google-cloud-dataflow-java-sdk-all</artifactId>
-      <version>0.3.150109</version>
+      <version>0.3.150326</version>
     </dependency>
 
     <!-- Google client dependencies -->
@@ -124,19 +124,6 @@
         </exclusion>
       </exclusions>
     </dependency>
-    <dependency>
-      <groupId>com.google.apis</groupId>
-      <artifactId>google-api-services-dataflow</artifactId>
-      <version>v1b4-rev1-1.19.1</version>
-      <exclusions>
-        <!-- Exclude an old version of guava which is being pulled
-             in by a transitive dependency google-api-client 1.19.0 -->
-        <exclusion>
-          <groupId>com.google.guava</groupId>
-          <artifactId>guava-jdk5</artifactId>
-        </exclusion>
-      </exclusions>
-    </dependency>
     <dependency>
       <groupId>com.google.cloud.genomics</groupId>
       <artifactId>google-genomics-utils</artifactId>

src/main/java/com/google/cloud/genomics/dataflow/pipelines/AnnotateVariants.java

@@ -250,8 +250,13 @@ private static String canonicalizeRefName(String refName) {
   }
 
   public static void main(String[] args) throws Exception {
+    // Register the options so that they show up via --help
+    PipelineOptionsFactory.register(GenomicsDatasetOptions.class);
     GenomicsDatasetOptions opts = PipelineOptionsFactory.fromArgs(args)
         .withValidation().as(GenomicsDatasetOptions.class);
+    // Option validation is not yet automatic, we make an explicit call here.
+    GenomicsDatasetOptions.Methods.validateOptions(opts);
+
     GenomicsFactory.OfflineAuth auth = GenomicsOptions.Methods.getGenomicsAuth(opts);
     Genomics genomics = auth.getGenomics(auth.getDefaultFactory());
 

src/main/java/com/google/cloud/genomics/dataflow/pipelines/CountReads.java

@@ -111,8 +111,12 @@ public Contig apply(String contigString) {
   }
 
   public static void main(String[] args) throws GeneralSecurityException, IOException {
+    // Register the options so that they show up via --help
+    PipelineOptionsFactory.register(CountReadsOptions.class);
     options = PipelineOptionsFactory.fromArgs(args).withValidation().as(CountReadsOptions.class);
-    GenomicsOptions.Methods.validateOptions(options);
+    // Option validation is not yet automatic, we make an explicit call here.
+    GenomicsDatasetOptions.Methods.validateOptions(options);
+
     auth = GCSOptions.Methods.createGCSAuth(options);
     p = Pipeline.create(options);
     DataflowWorkarounds.registerGenomicsCoders(p);
@@ -141,8 +145,8 @@ private static PCollection<Read> getReads() throws IOException {
 
   private static PCollection<Read> getReadsFromAPI() {
     List<SearchReadsRequest> requests = getReadRequests(options);
-    PCollection<SearchReadsRequest> readRequests =
-        DataflowWorkarounds.getPCollection(requests, p);
+    PCollection<SearchReadsRequest> readRequests = p.begin()
+        .apply(Create.of(requests));
     PCollection<Read> reads =
         readRequests.apply(
             ParDo.of(

src/main/java/com/google/cloud/genomics/dataflow/pipelines/IdentityByState.java

@@ -23,6 +23,7 @@
 import com.google.cloud.dataflow.sdk.io.TextIO;
 import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
 import com.google.cloud.dataflow.sdk.transforms.Combine;
+import com.google.cloud.dataflow.sdk.transforms.Create;
 import com.google.cloud.dataflow.sdk.transforms.ParDo;
 import com.google.cloud.dataflow.sdk.values.KV;
 import com.google.cloud.dataflow.sdk.values.PCollection;
@@ -52,8 +53,11 @@ public class IdentityByState {
 
   public static void main(String[] args) throws IOException, GeneralSecurityException,
       InstantiationException, IllegalAccessException {
+    // Register the options so that they show up via --help
+    PipelineOptionsFactory.register(IdentityByStateOptions.class);
     IdentityByStateOptions options =
         PipelineOptionsFactory.fromArgs(args).withValidation().as(IdentityByStateOptions.class);
+    // Option validation is not yet automatic, we make an explicit call here.
     GenomicsDatasetOptions.Methods.validateOptions(options);
 
     GenomicsFactory.OfflineAuth auth = GenomicsOptions.Methods.getGenomicsAuth(options);
@@ -63,8 +67,7 @@ public static void main(String[] args) throws IOException, GeneralSecurityExcept
 
     Pipeline p = Pipeline.create(options);
     DataflowWorkarounds.registerGenomicsCoders(p);
-    PCollection<SearchVariantsRequest> input =
-        DataflowWorkarounds.getPCollection(requests, p);
+    PCollection<SearchVariantsRequest> input = p.begin().apply(Create.of(requests));
 
     PCollection<Variant> variants =
         options.getHasNonVariantSegments()

src/main/java/com/google/cloud/genomics/dataflow/pipelines/TransmissionProbability.java

@@ -19,6 +19,7 @@
 import com.google.cloud.dataflow.sdk.Pipeline;
 import com.google.cloud.dataflow.sdk.io.TextIO;
 import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.transforms.Create;
 import com.google.cloud.dataflow.sdk.transforms.DoFn;
 import com.google.cloud.dataflow.sdk.transforms.GroupByKey;
 import com.google.cloud.dataflow.sdk.transforms.ParDo;
@@ -47,9 +48,12 @@ public class TransmissionProbability {
       = "nextPageToken,variants(id,start,names,calls(info,callSetName))";
 
   public static void main(String[] args) throws IOException, GeneralSecurityException {
+    // Register the options so that they show up via --help
+    PipelineOptionsFactory.register(GenomicsDatasetOptions.class);
     GenomicsDatasetOptions options = PipelineOptionsFactory.fromArgs(args)
         .withValidation().as(GenomicsDatasetOptions.class);
-    GenomicsOptions.Methods.validateOptions(options);
+    // Option validation is not yet automatic, we make an explicit call here.
+    GenomicsDatasetOptions.Methods.validateOptions(options);
 
     GenomicsFactory.OfflineAuth auth = GenomicsOptions.Methods.getGenomicsAuth(options);
     List<SearchVariantsRequest> requests = GenomicsDatasetOptions.Methods.getVariantRequests(
@@ -65,7 +69,8 @@ public static void main(String[] args) throws IOException, GeneralSecurityExcept
     //    - Groups Transmission sources by Variant,
     //    - Calculate transmission Probability for each variant
     //    - Print calculated values to a file.
-    DataflowWorkarounds.getPCollection(requests, p)
+    p.begin()
+        .apply(Create.of(requests))
         .apply(ParDo.named("VariantReader")
             .of(new VariantReader(auth, ShardBoundary.STRICT, VARIANT_FIELDS)))
         .apply(ParDo.named("ExtractFamilyVariantStatus")

src/main/java/com/google/cloud/genomics/dataflow/pipelines/VariantSimilarity.java

@@ -18,6 +18,7 @@
 import com.google.api.services.genomics.model.SearchVariantsRequest;
 import com.google.cloud.dataflow.sdk.Pipeline;
 import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.transforms.Create;
 import com.google.cloud.dataflow.sdk.transforms.ParDo;
 import com.google.cloud.genomics.dataflow.functions.ExtractSimilarCallsets;
 import com.google.cloud.genomics.dataflow.functions.OutputPCoAFile;
@@ -42,9 +43,12 @@ public class VariantSimilarity {
       = "nextPageToken,variants(start,calls(genotype,callSetName))";
 
   public static void main(String[] args) throws IOException, GeneralSecurityException {
+    // Register the options so that they show up via --help
+    PipelineOptionsFactory.register(GenomicsDatasetOptions.class);
     GenomicsDatasetOptions options = PipelineOptionsFactory.fromArgs(args)
         .withValidation().as(GenomicsDatasetOptions.class);
-    GenomicsOptions.Methods.validateOptions(options);
+    // Option validation is not yet automatic, we make an explicit call here.
+    GenomicsDatasetOptions.Methods.validateOptions(options);
 
     GenomicsFactory.OfflineAuth auth = GenomicsOptions.Methods.getGenomicsAuth(options);
     List<SearchVariantsRequest> requests =
@@ -53,8 +57,8 @@ public static void main(String[] args) throws IOException, GeneralSecurityExcept
 
     Pipeline p = Pipeline.create(options);
     DataflowWorkarounds.registerGenomicsCoders(p);
-    DataflowWorkarounds
-        .getPCollection(requests, p)
+    p.begin()
+        .apply(Create.of(requests))
         .apply(
             ParDo.named("VariantReader").of(
                 new VariantReader(auth, ShardBoundary.STRICT, VARIANT_FIELDS)))

src/main/java/com/google/cloud/genomics/dataflow/readers/bam/ReadBAMTransform.java

@@ -69,10 +69,10 @@ public static PCollection<Read> getReadsFromBAMFilesSharded(
 
   public static class ShardFn extends DoFn<String, BAMShard> {
     GenomicsFactory.OfflineAuth auth;
-    PCollectionView<Iterable<Contig>, ?> contigsView;
+    PCollectionView<Iterable<Contig>> contigsView;
     Storage.Objects storage;
 
-    public ShardFn(GenomicsFactory.OfflineAuth auth, PCollectionView<Iterable<Contig>, ?> contigsView) {
+    public ShardFn(GenomicsFactory.OfflineAuth auth, PCollectionView<Iterable<Contig>> contigsView) {
       this.auth = auth;
       this.contigsView = contigsView;
     }
@@ -115,7 +115,7 @@ public void processElement(ProcessContext c) throws java.lang.Exception {
   public PCollection<Read> apply(PCollectionTuple contigsAndBAMs) {
 
     final PCollection<Contig> contigs = contigsAndBAMs.get(CONTIGS_TAG);
-    final PCollectionView<Iterable<Contig>, ?> contigsView =
+    final PCollectionView<Iterable<Contig>> contigsView =
         contigs.apply(View.<Contig>asIterable());
 
     final PCollection<String> BAMFileGCSPaths = contigsAndBAMs.get(BAMS_TAG);

src/main/java/com/google/cloud/genomics/dataflow/utils/DataflowWorkarounds.java

@@ -15,23 +15,13 @@
  */
 package com.google.cloud.genomics.dataflow.utils;
 
-import com.google.api.client.json.GenericJson;
-import com.google.cloud.dataflow.sdk.Pipeline;
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.CoderRegistry;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
-import com.google.cloud.dataflow.sdk.options.PipelineOptions;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRegistrar.Options;
-import com.google.cloud.dataflow.sdk.transforms.Create;
-import com.google.cloud.dataflow.sdk.transforms.Flatten;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.cloud.dataflow.sdk.values.PCollectionList;
-import com.google.cloud.genomics.dataflow.coders.GenericJsonCoder;
-import com.google.common.base.Predicate;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.common.primitives.UnsignedInts;
+import static com.google.common.collect.Lists.newArrayList;
+
+import java.net.URL;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.logging.Logger;
 
 import org.reflections.Reflections;
 import org.reflections.scanners.ResourcesScanner;
@@ -40,13 +30,13 @@
 import org.reflections.util.ConfigurationBuilder;
 import org.reflections.util.FilterBuilder;
 
-import java.net.URL;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.logging.Logger;
-
-import static com.google.common.collect.Lists.newArrayList;
+import com.google.api.client.json.GenericJson;
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.coders.Coder;
+import com.google.cloud.dataflow.sdk.coders.CoderRegistry;
+import com.google.cloud.genomics.dataflow.coders.GenericJsonCoder;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
 
 /**
  * Contains dataflow-related workarounds.
@@ -107,61 +97,4 @@ public boolean apply(URL url) {
       DataflowWorkarounds.registerCoder(p, clazz, GenericJsonCoder.of(clazz));
     }
   }
-  
-  /**
-   * Change a flat list of sharding options into a flattened PCollection to force dataflow to use
-   * multiple workers. In the future, this shouldn't be necessary.
-   */
-  public static <T> PCollection<T> getPCollection(List<T> shardOptions, Coder<T> coder,
-      Pipeline p) {
-    
-    DataflowPipelineOptions options = (GenomicsOptions) p.getOptions();
-    int numWorkers = options.getNumWorkers();
-    String machineType = options.getMachineType();
-    
-    if (machineType != null) {
-      String[] machineNameParts =
-          Iterables.toArray(Splitter.on('-').split(machineType), String.class);
-      if (3 == machineNameParts.length) {
-        try {
-          int numCores = UnsignedInts.parseUnsignedInt(machineNameParts[2]);
-          numWorkers *= numCores;
-        } catch (Exception e) {
-          LOG.warning("Assuming one core per worker: " + e);
-        }
-      }
-    }
-
-    LOG.info("Turning " + shardOptions.size() + " options into " + numWorkers + " workers");
-    numWorkers = Math.min(shardOptions.size(), numWorkers);
-
-    int optionsPerWorker = (int) Math.ceil(shardOptions.size() / numWorkers);
-    List<PCollection<T>> pCollections = Lists.newArrayList();
-
-    
-    for (int i = 0; i < numWorkers; i++) {
-      int start = i * optionsPerWorker;
-      int end = Math.min(shardOptions.size(), start + optionsPerWorker);
-      
-      // It's possible for start >= end in the last worker,
-      // in which case we'll just skip the collection.
-      if (start >= end) {
-        break;
-      }
-      
-      LOG.info("Adding collection with " + start + " to " + end);
-      PCollection<T> collection = p.begin().apply(Create.of(shardOptions.subList(start, end)));
-      if (coder != null) {
-        collection.setCoder(coder);
-      }
-      pCollections.add(collection);
-    }
-
-    return PCollectionList.of(pCollections).apply(Flatten.<T>create());
-  }
-  
-  public static <T> PCollection<T> getPCollection(
-      List<T> shardOptions, Pipeline p) {
-    return getPCollection(shardOptions, null, p);
-  }
 }