{"action":"created","comment":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments/35792130","id":35792130,"diff_hunk":"@@ -0,0 +1,436 @@\n+package com.google.cloud.dataflow.contrib.hadoop;\n+\n+import com.google.cloud.dataflow.sdk.coders.Coder;\n+import com.google.cloud.dataflow.sdk.coders.KvCoder;\n+import com.google.cloud.dataflow.sdk.coders.VoidCoder;\n+import com.google.cloud.dataflow.sdk.io.BoundedSource;\n+import com.google.cloud.dataflow.sdk.options.PipelineOptions;\n+import com.google.cloud.dataflow.sdk.util.ExecutionContext;\n+import com.google.cloud.dataflow.sdk.values.KV;\n+import com.google.common.base.Function;\n+import com.google.common.base.Preconditions;\n+import com.google.common.collect.ImmutableList;\n+import com.google.common.collect.Lists;\n+import java.io.Externalizable;\n+import java.io.IOException;\n+import java.io.ObjectInput;\n+import java.io.ObjectOutput;\n+import java.lang.reflect.InvocationTargetException;\n+import java.lang.reflect.Method;\n+import java.util.List;\n+import java.util.ListIterator;\n+import java.util.NoSuchElementException;\n+import javax.annotation.Nullable;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.FileStatus;\n+import org.apache.hadoop.fs.Path;\n+import org.apache.hadoop.io.Writable;\n+import org.apache.hadoop.io.WritableUtils;\n+import org.apache.hadoop.mapreduce.InputSplit;\n+import org.apache.hadoop.mapreduce.Job;\n+import org.apache.hadoop.mapreduce.JobContext;\n+import org.apache.hadoop.mapreduce.RecordReader;\n+import org.apache.hadoop.mapreduce.TaskAttemptContext;\n+import org.apache.hadoop.mapreduce.TaskAttemptID;\n+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;\n+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;\n+\n+/**\n+ * A {@code BoundedSource} for reading files resident in a Hadoop filesystem using a\n+ * Hadoop file-based input format.\n+ *\n+ * <p>To read a {@link com.google.cloud.dataflow.sdk.values.PCollection} of\n+ * {@link com.google.cloud.dataflow.sdk.values.KV} key-value pairs from one or more\n+ * Hadoop files, use {@link HadoopFileSource#from} to specify the path(s) of the files to\n+ * read, the Hadoop {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}, the\n+ * key class and the value class.\n+ *\n+ * <p>A {@code HadoopFileSource} can be read from using the\n+ * {@link com.google.cloud.dataflow.sdk.io.Read} transform. For example:\n+ *\n+ * <pre>\n+ * {@code\n+ * HadoopFileSource<K, V> source = HadoopFileSource.from(path, MyInputFormat.class,\n+ *   MyKey.class, MyValue.class);\n+ * PCollection<KV<MyKey, MyValue>> records = Read.from(mySource);\n+ * }\n+ * </pre>\n+ *\n+ * Implementation note: Since Hadoop's {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}\n+ * determines the input splits, this class extends {@link BoundedSource} rather than\n+ * {@link com.google.cloud.dataflow.sdk.io.ByteOffsetBasedSource}, since the latter\n+ * dictates input splits.\n+\n+ * @param <K> The type of keys to be read from the source.\n+ * @param <V> The type of values to be read from the source.\n+ */\n+public class HadoopFileSource<K, V> extends BoundedSource<KV<K, V>> {\n+\n+  private final String filepattern;\n+  private final Class<? extends FileInputFormat<?, ?>> formatClass;\n+  private final Class<K> keyClass;\n+  private final Class<V> valueClass;\n+  private final SerializableSplit serializableSplit;\n+\n+  /**\n+   * Creates a {@code HadoopFileSource} that reads from the given file name or pattern (\"glob\")\n+   * using the given Hadoop {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat},\n+   * with key-value types specified by the given key class and value class.\n+   */\n+  public static <K, V, F extends FileInputFormat<K, V>> HadoopFileSource<K, V> from(String filepattern,\n+      Class<F> formatClass, Class<K> keyClass, Class<V> valueClass) {\n+    return new HadoopFileSource(filepattern, formatClass, keyClass, valueClass);\n+  }\n+\n+  /**\n+   * Create a {@code HadoopFileSource} based on a file or a file pattern specification.\n+   */\n+  private HadoopFileSource(String filepattern,\n+      Class<? extends FileInputFormat<?, ?>> formatClass, Class<K> keyClass,\n+      Class<V> valueClass) {\n+    this(filepattern, formatClass, keyClass, valueClass, null);\n+  }\n+\n+  /**\n+   * Create a {@code HadoopFileSource} based on a single Hadoop input split, which won't be\n+   * split up further.\n+   */\n+  private HadoopFileSource(String filepattern,\n+      Class<? extends FileInputFormat<?, ?>> formatClass, Class<K> keyClass,\n+      Class<V> valueClass, SerializableSplit serializableSplit) {\n+    this.filepattern = filepattern;\n+    this.formatClass = formatClass;\n+    this.keyClass = keyClass;\n+    this.valueClass = valueClass;\n+    this.serializableSplit = serializableSplit;\n+  }\n+\n+  public String getFilepattern() {\n+    return filepattern;\n+  }\n+\n+  public Class<? extends FileInputFormat<?, ?>> getFormatClass() {\n+    return formatClass;\n+  }\n+\n+  public Class<K> getKeyClass() {\n+    return keyClass;\n+  }\n+\n+  public Class<V> getValueClass() {\n+    return valueClass;\n+  }\n+\n+  @Override\n+  public void validate() {\n+    Preconditions.checkNotNull(filepattern,\n+        \"need to set the filepattern of a HadoopFileSource\");\n+    Preconditions.checkNotNull(formatClass,\n+        \"need to set the format class of a HadoopFileSource\");\n+    Preconditions.checkNotNull(keyClass,\n+        \"need to set the key class of a HadoopFileSource\");\n+    Preconditions.checkNotNull(valueClass,\n+        \"need to set the value class of a HadoopFileSource\");\n+  }\n+\n+  @Override\n+  public List<? extends BoundedSource<KV<K, V>>> splitIntoBundles(long desiredBundleSizeBytes,\n+      PipelineOptions options) throws Exception {\n+    if (serializableSplit == null) {\n+      return Lists.transform(computeSplits(desiredBundleSizeBytes),\n+          new Function<InputSplit, BoundedSource<KV<K, V>>>() {\n+        @Nullable @Override\n+        public BoundedSource<KV<K, V>> apply(@Nullable InputSplit inputSplit) {\n+          return new HadoopFileSource<K, V>(filepattern, formatClass, keyClass,\n+              valueClass, new SerializableSplit(inputSplit));\n+        }\n+      });\n+    } else {\n+      return ImmutableList.of(this);\n+    }\n+  }\n+\n+  private FileInputFormat<?, ?> createFormat(Job job) throws IOException, IllegalAccessException,\n+      InstantiationException {\n+    Path path = new Path(filepattern);\n+    FileInputFormat.addInputPath(job, path);\n+    return formatClass.newInstance();\n+  }\n+\n+  private List<InputSplit> computeSplits(long desiredBundleSizeBytes) throws IOException, IllegalAccessException,\n+      InstantiationException {\n+    Job job = Job.getInstance();\n+    FileInputFormat.setMinInputSplitSize(job, desiredBundleSizeBytes);\n+    FileInputFormat.setMaxInputSplitSize(job, desiredBundleSizeBytes);\n+    return createFormat(job).getSplits(job);\n+  }\n+\n+  @Override\n+  public BoundedReader<KV<K, V>> createReader(PipelineOptions options, @Nullable\n+      ExecutionContext executionContext) throws IOException {\n+    this.validate();\n+\n+    if (serializableSplit == null) {\n+      return new HadoopFileReader<>(this, filepattern, formatClass);\n+    } else {\n+      return new HadoopFileReader<>(this, filepattern, formatClass,\n+          serializableSplit.getSplit());\n+    }\n+  }\n+\n+  @Override\n+  public Coder<KV<K, V>> getDefaultOutputCoder() {\n+    return KvCoder.of(getDefaultCoder(keyClass), getDefaultCoder(valueClass));\n+  }\n+\n+  @SuppressWarnings(\"unchecked\")\n+  private <T> Coder<T> getDefaultCoder(Class<T> c) {\n+    if (Writable.class.isAssignableFrom(c)) {\n+      Class<? extends Writable> writableClass = (Class<? extends Writable>) c;\n+      return (Coder<T>) WritableCoder.of(writableClass);\n+    } else if (Void.class.equals(c)) {\n+      return (Coder<T>) VoidCoder.of();\n+    }\n+    // TODO: how to use registered coders here?\n+    throw new IllegalStateException(\"Cannot find coder for \" + c);\n+  }\n+\n+  // BoundedSource\n+\n+  @Override\n+  public long getEstimatedSizeBytes(PipelineOptions options) {\n+    long size = 0;\n+    try {\n+      Job job = Job.getInstance();\n+      for (FileStatus st : listStatus(createFormat(job), job)) {\n+        size += st.getLen();\n+      }\n+    } catch (IOException | NoSuchMethodException | InvocationTargetException |\n+        IllegalAccessException | InstantiationException e) {\n+      // ignore, and return 0\n+    }\n+    return size;\n+  }\n+\n+  private <K, V> List<FileStatus> listStatus(FileInputFormat<K, V> format,\n+      JobContext jobContext) throws NoSuchMethodException, InvocationTargetException,\n+      IllegalAccessException {\n+    // FileInputFormat#listStatus is protected, so call using reflection\n+    Method listStatus = FileInputFormat.class.getDeclaredMethod(\"listStatus\", JobContext.class);\n+    listStatus.setAccessible(true);\n+    @SuppressWarnings(\"unchecked\")\n+    List<FileStatus> stat = (List<FileStatus>) listStatus.invoke(format, jobContext);\n+    return stat;\n+  }\n+\n+  @Override\n+  public boolean producesSortedKeys(PipelineOptions options) throws Exception {\n+    return false;\n+  }\n+\n+  static class HadoopFileReader<K, V> extends BoundedSource.AbstractBoundedReader<KV<K, V>> {\n+\n+    private final BoundedSource<KV<K, V>> source;\n+    private final String filepattern;\n+    private final Class formatClass;\n+\n+    private FileInputFormat<?, ?> format;\n+    private TaskAttemptContext attemptContext;\n+    private List<InputSplit> splits;\n+    private ListIterator<InputSplit> splitsIterator;\n+    private Configuration conf;\n+    private RecordReader<K, V> currentReader;\n+    private KV<K, V> currentPair;\n+\n+    /**\n+     * Create a {@code HadoopFileReader} based on a file or a file pattern specification.\n+     */\n+    public HadoopFileReader(BoundedSource<KV<K, V>> source, String filepattern,\n+        Class<? extends FileInputFormat<?, ?>> formatClass) {\n+      this(source, filepattern, formatClass, null);\n+    }\n+\n+    /**\n+     * Create a {@code HadoopFileReader} based on a single Hadoop input split.\n+     */\n+    public HadoopFileReader(BoundedSource<KV<K, V>> source, String filepattern,\n+        Class<? extends FileInputFormat<?, ?>> formatClass, InputSplit split) {\n+      this.source = source;\n+      this.filepattern = filepattern;\n+      this.formatClass = formatClass;\n+      if (split != null) {\n+        this.splits = ImmutableList.of(split);\n+        this.splitsIterator = splits.listIterator();\n+      }\n+    }\n+\n+    @Override\n+    public boolean start() throws IOException {\n+      Job job = Job.getInstance();\n+      Path path = new Path(filepattern);\n+      FileInputFormat.addInputPath(job, path);\n+\n+      try {\n+        @SuppressWarnings(\"unchecked\")\n+        FileInputFormat<K, V> f = (FileInputFormat<K, V>) formatClass.newInstance();\n+        this.format = f;\n+      } catch (InstantiationException | IllegalAccessException e) {\n+        throw new IOException(\"Cannot instantiate file input format \" + formatClass, e);\n+      }\n+      this.attemptContext = new TaskAttemptContextImpl(job.getConfiguration(),\n+          new TaskAttemptID());\n+\n+      if (splitsIterator == null) {\n+        this.splits = format.getSplits(job);\n+        this.splitsIterator = splits.listIterator();\n+      }\n+      this.conf = job.getConfiguration();\n+      return advance();\n+    }\n+\n+    @Override\n+    public boolean advance() throws IOException {\n+      try {\n+        if (currentReader != null && currentReader.nextKeyValue()) {\n+          currentPair = nextPair();\n+          return true;\n+        } else {\n+          while (splitsIterator.hasNext()) {\n+            // advance the reader and see if it has records\n+            InputSplit nextSplit = splitsIterator.next();\n+            @SuppressWarnings(\"unchecked\")\n+            RecordReader<K, V> reader = (RecordReader<K, V>) format.createRecordReader(nextSplit, attemptContext);\n+            currentReader = reader;\n+            currentReader.initialize(nextSplit, attemptContext);\n+            if (currentReader.nextKeyValue()) {\n+              currentPair = nextPair();\n+              return true;\n+            }\n+          }\n+          // either no next split or all readers were empty\n+          currentPair = null;\n+          return false;\n+        }\n+      } catch (InterruptedException e) {\n+        Thread.currentThread().interrupt();\n+        throw new IOException(e);\n+      }\n+    }\n+\n+    @SuppressWarnings(\"unchecked\")\n+    private KV<K, V> nextPair() throws IOException, InterruptedException {\n+      K key = currentReader.getCurrentKey();\n+      V value = currentReader.getCurrentValue();\n+      if (key instanceof Writable) {\n+        key = (K) WritableUtils.clone((Writable) key, conf);\n+      }\n+      if (value instanceof Writable) {\n+        value = (V) WritableUtils.clone((Writable) value, conf);\n+      }\n+      return KV.of(key, value);\n+    }\n+\n+    @Override\n+    public KV<K, V> getCurrent() throws NoSuchElementException {\n+      if (currentPair == null) {\n+        throw new NoSuchElementException();\n+      }\n+      return currentPair;\n+    }\n+\n+    @Override\n+    public void close() throws IOException {\n+      if (currentReader != null) {\n+        currentReader.close();\n+      }\n+      currentPair = null;\n+    }\n+\n+    @Override\n+    public BoundedSource<KV<K, V>> getCurrentSource() {\n+      return source;\n+    }\n+\n+    // BoundedReader\n+\n+    @Override\n+    public Double getFractionConsumed() {\n+      if (currentReader == null) {\n+        return 0.0;\n+      }\n+      if (splits.isEmpty()) {\n+        return 1.0;\n+      }\n+      int index = splitsIterator.previousIndex();\n+      int numReaders = splits.size();\n+      if (index == numReaders) {\n+        return 1.0;\n+      }\n+      double before = 1.0 * index / numReaders;\n+      double after = 1.0 * (index + 1) / numReaders;\n+      Double fractionOfCurrentReader = getProgress();\n+      if (fractionOfCurrentReader == null) {\n+        return before;\n+      }\n+      return before + fractionOfCurrentReader * (after - before);\n+    }\n+\n+    private Double getProgress() {\n+      try {\n+        return (double) currentReader.getProgress();\n+      } catch (IOException e) {\n+        return null;\n+      } catch (InterruptedException e) {\n+        return null;\n+      }\n+    }\n+\n+    @Override\n+    public BoundedSource<KV<K, V>> splitAtFraction(double fraction) {\n+      // not supported","path":"contrib/hadoop/src/main/java/com/google/cloud/dataflow/contrib/hadoop/HadoopFileSource.java","position":390,"original_position":390,"commit_id":"e47136ea4c5f37fd9efdac30e725002021c801cc","original_commit_id":"e47136ea4c5f37fd9efdac30e725002021c801cc","user":{"login":"jkff","id":343722,"avatar_url":"https://avatars.githubusercontent.com/u/343722?v=3","gravatar_id":"","url":"https://api.github.com/users/jkff","html_url":"https://github.com/jkff","followers_url":"https://api.github.com/users/jkff/followers","following_url":"https://api.github.com/users/jkff/following{/other_user}","gists_url":"https://api.github.com/users/jkff/gists{/gist_id}","starred_url":"https://api.github.com/users/jkff/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jkff/subscriptions","organizations_url":"https://api.github.com/users/jkff/orgs","repos_url":"https://api.github.com/users/jkff/repos","events_url":"https://api.github.com/users/jkff/events{/privacy}","received_events_url":"https://api.github.com/users/jkff/received_events","type":"User","site_admin":false},"body":"Explain in a bit more detail why it is not supported? (if it's a fundamental limitation of Hadoop file API, explain which one - e.g. so that somebody who later wants to try adding this support is aware of which limitation they're up against)","created_at":"2015-07-29T18:10:25Z","updated_at":"2015-07-29T18:10:25Z","html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47#discussion_r35792130","pull_request_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47","_links":{"self":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments/35792130"},"html":{"href":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47#discussion_r35792130"},"pull_request":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47"}}},"pull_request":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47","id":40694776,"html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47","diff_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47.diff","patch_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47.patch","issue_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/47","number":47,"state":"open","locked":false,"title":"Add contrib module for HadoopFileSource.","user":{"login":"tomwhite","id":85085,"avatar_url":"https://avatars.githubusercontent.com/u/85085?v=3","gravatar_id":"","url":"https://api.github.com/users/tomwhite","html_url":"https://github.com/tomwhite","followers_url":"https://api.github.com/users/tomwhite/followers","following_url":"https://api.github.com/users/tomwhite/following{/other_user}","gists_url":"https://api.github.com/users/tomwhite/gists{/gist_id}","starred_url":"https://api.github.com/users/tomwhite/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tomwhite/subscriptions","organizations_url":"https://api.github.com/users/tomwhite/orgs","repos_url":"https://api.github.com/users/tomwhite/repos","events_url":"https://api.github.com/users/tomwhite/events{/privacy}","received_events_url":"https://api.github.com/users/tomwhite/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-07-23T17:06:40Z","updated_at":"2015-07-29T18:10:25Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f835219c1c6998b80651748851326b0c177d7593","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47/commits","review_comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47/comments","review_comment_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments{/number}","comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/47/comments","statuses_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/statuses/e47136ea4c5f37fd9efdac30e725002021c801cc","head":{"label":"tomwhite:hadoop-file-source","ref":"hadoop-file-source","sha":"e47136ea4c5f37fd9efdac30e725002021c801cc","user":{"login":"tomwhite","id":85085,"avatar_url":"https://avatars.githubusercontent.com/u/85085?v=3","gravatar_id":"","url":"https://api.github.com/users/tomwhite","html_url":"https://github.com/tomwhite","followers_url":"https://api.github.com/users/tomwhite/followers","following_url":"https://api.github.com/users/tomwhite/following{/other_user}","gists_url":"https://api.github.com/users/tomwhite/gists{/gist_id}","starred_url":"https://api.github.com/users/tomwhite/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tomwhite/subscriptions","organizations_url":"https://api.github.com/users/tomwhite/orgs","repos_url":"https://api.github.com/users/tomwhite/repos","events_url":"https://api.github.com/users/tomwhite/events{/privacy}","received_events_url":"https://api.github.com/users/tomwhite/received_events","type":"User","site_admin":false},"repo":{"id":37258553,"name":"DataflowJavaSDK","full_name":"tomwhite/DataflowJavaSDK","owner":{"login":"tomwhite","id":85085,"avatar_url":"https://avatars.githubusercontent.com/u/85085?v=3","gravatar_id":"","url":"https://api.github.com/users/tomwhite","html_url":"https://github.com/tomwhite","followers_url":"https://api.github.com/users/tomwhite/followers","following_url":"https://api.github.com/users/tomwhite/following{/other_user}","gists_url":"https://api.github.com/users/tomwhite/gists{/gist_id}","starred_url":"https://api.github.com/users/tomwhite/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tomwhite/subscriptions","organizations_url":"https://api.github.com/users/tomwhite/orgs","repos_url":"https://api.github.com/users/tomwhite/repos","events_url":"https://api.github.com/users/tomwhite/events{/privacy}","received_events_url":"https://api.github.com/users/tomwhite/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tomwhite/DataflowJavaSDK","description":"Google Cloud Dataflow provides a simple, powerful model for building both batch and streaming parallel data processing pipelines.","fork":true,"url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK","forks_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/forks","keys_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/teams","hooks_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/hooks","issue_events_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/issues/events{/number}","events_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/events","assignees_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/assignees{/user}","branches_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/branches{/branch}","tags_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/tags","blobs_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/git/refs{/sha}","trees_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/statuses/{sha}","languages_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/languages","stargazers_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/stargazers","contributors_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/contributors","subscribers_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/subscribers","subscription_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/subscription","commits_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/commits{/sha}","git_commits_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/git/commits{/sha}","comments_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/comments{/number}","issue_comment_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/issues/comments{/number}","contents_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/contents/{+path}","compare_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/merges","archive_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/downloads","issues_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/issues{/number}","pulls_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/pulls{/number}","milestones_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/milestones{/number}","notifications_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/labels{/name}","releases_url":"https://api.github.com/repos/tomwhite/DataflowJavaSDK/releases{/id}","created_at":"2015-06-11T12:02:14Z","updated_at":"2015-06-11T12:02:16Z","pushed_at":"2015-07-23T17:05:19Z","git_url":"git://github.com/tomwhite/DataflowJavaSDK.git","ssh_url":"git@github.com:tomwhite/DataflowJavaSDK.git","clone_url":"https://github.com/tomwhite/DataflowJavaSDK.git","svn_url":"https://github.com/tomwhite/DataflowJavaSDK","homepage":"http://cloud.google.com/dataflow","size":4294,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":false,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"GoogleCloudPlatform:master","ref":"master","sha":"4e3bc5d580c310f7eea46985eb986ab7cc3aea61","user":{"login":"GoogleCloudPlatform","id":2810941,"avatar_url":"https://avatars.githubusercontent.com/u/2810941?v=3","gravatar_id":"","url":"https://api.github.com/users/GoogleCloudPlatform","html_url":"https://github.com/GoogleCloudPlatform","followers_url":"https://api.github.com/users/GoogleCloudPlatform/followers","following_url":"https://api.github.com/users/GoogleCloudPlatform/following{/other_user}","gists_url":"https://api.github.com/users/GoogleCloudPlatform/gists{/gist_id}","starred_url":"https://api.github.com/users/GoogleCloudPlatform/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/GoogleCloudPlatform/subscriptions","organizations_url":"https://api.github.com/users/GoogleCloudPlatform/orgs","repos_url":"https://api.github.com/users/GoogleCloudPlatform/repos","events_url":"https://api.github.com/users/GoogleCloudPlatform/events{/privacy}","received_events_url":"https://api.github.com/users/GoogleCloudPlatform/received_events","type":"Organization","site_admin":false},"repo":{"id":27503425,"name":"DataflowJavaSDK","full_name":"GoogleCloudPlatform/DataflowJavaSDK","owner":{"login":"GoogleCloudPlatform","id":2810941,"avatar_url":"https://avatars.githubusercontent.com/u/2810941?v=3","gravatar_id":"","url":"https://api.github.com/users/GoogleCloudPlatform","html_url":"https://github.com/GoogleCloudPlatform","followers_url":"https://api.github.com/users/GoogleCloudPlatform/followers","following_url":"https://api.github.com/users/GoogleCloudPlatform/following{/other_user}","gists_url":"https://api.github.com/users/GoogleCloudPlatform/gists{/gist_id}","starred_url":"https://api.github.com/users/GoogleCloudPlatform/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/GoogleCloudPlatform/subscriptions","organizations_url":"https://api.github.com/users/GoogleCloudPlatform/orgs","repos_url":"https://api.github.com/users/GoogleCloudPlatform/repos","events_url":"https://api.github.com/users/GoogleCloudPlatform/events{/privacy}","received_events_url":"https://api.github.com/users/GoogleCloudPlatform/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK","description":"Google Cloud Dataflow provides a simple, powerful model for building both batch and streaming parallel data processing pipelines.","fork":false,"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK","forks_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/forks","keys_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/keys{/key_id}","collaborators_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/teams","hooks_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/hooks","issue_events_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/events{/number}","events_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/events","assignees_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/assignees{/user}","branches_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/branches{/branch}","tags_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/tags","blobs_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/git/refs{/sha}","trees_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/git/trees{/sha}","statuses_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/statuses/{sha}","languages_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/languages","stargazers_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/stargazers","contributors_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/contributors","subscribers_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/subscribers","subscription_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/subscription","commits_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/commits{/sha}","git_commits_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/git/commits{/sha}","comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/comments{/number}","issue_comment_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/comments{/number}","contents_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/contents/{+path}","compare_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/compare/{base}...{head}","merges_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/merges","archive_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/downloads","issues_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues{/number}","pulls_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls{/number}","milestones_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/milestones{/number}","notifications_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/labels{/name}","releases_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/releases{/id}","created_at":"2014-12-03T19:20:21Z","updated_at":"2015-07-28T16:40:25Z","pushed_at":"2015-07-28T17:18:09Z","git_url":"git://github.com/GoogleCloudPlatform/DataflowJavaSDK.git","ssh_url":"git@github.com:GoogleCloudPlatform/DataflowJavaSDK.git","clone_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK.git","svn_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK","homepage":"http://cloud.google.com/dataflow","size":5950,"stargazers_count":329,"watchers_count":329,"language":"Java","has_issues":true,"has_downloads":false,"has_wiki":false,"has_pages":false,"forks_count":95,"mirror_url":null,"open_issues_count":17,"forks":95,"open_issues":17,"watchers":329,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47"},"html":{"href":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/47"},"issue":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/47"},"comments":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/47/comments"},"review_comments":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47/comments"},"review_comment":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/47/commits"},"statuses":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/statuses/e47136ea4c5f37fd9efdac30e725002021c801cc"}}}}