{"action":"created","comment":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments/56732321","id":56732321,"diff_hunk":"@@ -0,0 +1,257 @@\n+/*\n+ * Copyright (C) 2015 Google Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\"); you may not\n+ * use this file except in compliance with the License. You may obtain a copy of\n+ * the License at\n+ *\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n+ * License for the specific language governing permissions and limitations under\n+ * the License.\n+ */\n+\n+package com.google.cloud.dataflow.contrib.kafka.examples;\n+\n+import com.google.cloud.dataflow.contrib.kafka.KafkaIO;\n+import com.google.cloud.dataflow.sdk.Pipeline;\n+import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;\n+import com.google.cloud.dataflow.sdk.options.Default;\n+import com.google.cloud.dataflow.sdk.options.Description;\n+import com.google.cloud.dataflow.sdk.options.PipelineOptions;\n+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;\n+import com.google.cloud.dataflow.sdk.options.Validation.Required;\n+import com.google.cloud.dataflow.sdk.transforms.Count;\n+import com.google.cloud.dataflow.sdk.transforms.DoFn;\n+import com.google.cloud.dataflow.sdk.transforms.ParDo;\n+import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;\n+import com.google.cloud.dataflow.sdk.transforms.Top;\n+import com.google.cloud.dataflow.sdk.transforms.Values;\n+import com.google.cloud.dataflow.sdk.transforms.windowing.IntervalWindow;\n+import com.google.cloud.dataflow.sdk.transforms.windowing.SlidingWindows;\n+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;\n+import com.google.cloud.dataflow.sdk.values.KV;\n+import com.google.common.base.Throwables;\n+import com.google.common.collect.ImmutableMap;\n+\n+import com.fasterxml.jackson.annotation.JsonProperty;\n+import com.fasterxml.jackson.databind.JsonNode;\n+import com.fasterxml.jackson.databind.ObjectMapper;\n+import com.fasterxml.jackson.databind.ObjectWriter;\n+\n+import org.apache.kafka.clients.producer.KafkaProducer;\n+import org.apache.kafka.clients.producer.ProducerRecord;\n+import org.apache.kafka.common.serialization.StringSerializer;\n+import org.joda.time.Duration;\n+import org.joda.time.Instant;\n+import org.joda.time.format.DateTimeFormat;\n+import org.joda.time.format.DateTimeFormatter;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.util.ArrayList;\n+import java.util.List;\n+import java.util.Map;\n+\n+/**\n+ * This Dataflow app show cases {@link KafkaIO}. The application reads from a Kafka topic\n+ * containing <a href=\"https://dev.twitter.com/overview/api/tweets\">JSON Tweets</a>, calculates top\n+ * hashtags in 10 minute window. The results are written back to a Kafka topic.\n+ *\n+ * <pre>{@code\n+ * Usage:\n+ *   $ java -cp jar_with_dependencies.jar                                           \\\n+ *          com.google.cloud.dataflow.contrib.kafka.examples.TopHashtagsExample     \\\n+ *          --project=GCP_PROJECT                                                   \\\n+ *          --stagingLocation=GS_STAGING_DIRECTORY                                  \\\n+ *          --runner=BlockingDataflowPipelineRunner                                 \\\n+ *          --bootstrapServers=\"kafka_server_1:9092\"                                \\\n+ *          --topics=\"sample_tweets_json\"                                           \\\n+ *          --outputTopic=\"top_hashtags\"\n+ * }</pre>\n+ */\n+public class TopHashtagsExample {\n+\n+  private static final Logger LOG = LoggerFactory.getLogger(TopHashtagsExample.class);\n+\n+  /**\n+   * Options for the app.\n+   */\n+  public static interface Options extends PipelineOptions {\n+    @Description(\"Sliding window length in minutes\")\n+    @Default.Integer(10)\n+    Integer getSlidingWindowLengthMinutes();\n+    void setSlidingWindowLengthMinutes(Integer value);\n+\n+    @Description(\"Trigger window interval in minutes\")\n+    @Default.Integer(1)\n+    Integer getSlidingWindowIntervalMinutes();\n+    void setSlidingWindowIntervalMinutes(Integer value);\n+\n+    @Description(\"Bootstrap Server(s) for Kafka\")\n+    @Required\n+    String getBootstrapServers();\n+    void setBootstrapServers(String servers);\n+\n+    @Description(\"One or more comma seperated topics to read from\")\n+    @Required\n+    List<String> getTopics();\n+    void setTopics(List<String> topics);\n+\n+    @Description(\"Number of Top Hashtags to track\")\n+    @Default.Integer(10)\n+    Integer getNumTopHashtags();\n+    void setNumTopHashtags(Integer count);\n+\n+    @Description(\"Kafka topic name for writing results\")\n+    @Required\n+    String getOutputTopic();\n+    void setOutputTopic(String topic);\n+  }\n+\n+  public static void main(String args[]) {\n+\n+    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);\n+\n+    Pipeline pipeline = Pipeline.create(options);\n+\n+    KafkaIO.Read<?, String> reader = KafkaIO.read()\n+        .withBootstrapServers(options.getBootstrapServers())\n+        .withTopics(options.getTopics())\n+        .withValueCoder(StringUtf8Coder.of())\n+        .withTimestampFn(timestampFn);\n+\n+    pipeline\n+      .apply(\"sample_tweets\", reader)\n+      .apply(Values.<String>create())\n+      .apply(ParDo.of(new ExtractHashtagsFn()))\n+      .apply(Window.<String>into(SlidingWindows\n+          .of(Duration.standardMinutes(options.getSlidingWindowLengthMinutes()))\n+          .every(Duration.standardMinutes(options.getSlidingWindowIntervalMinutes()))))\n+      .apply(Count.<String>perElement())\n+      .apply(Top.of(options.getNumTopHashtags(), new KV.OrderByValue<String, Long>())\n+                .withoutDefaults())\n+      .apply(ParDo.of(new OutputFormatter()))\n+      .apply(ParDo.of(new KafkaWriter(options)));\n+\n+    pipeline.run();\n+  }\n+\n+  // The rest of the file implements DoFns to extract hashtags, formatting output, writing output\n+  // back to Kafka. Note that writing to Kafka is not a complete Dataflow Sink. It is a best-effort\n+  // logging of the results.\n+\n+  private static final ObjectMapper jsonMapper = new ObjectMapper();\n+\n+  /**\n+   * Emit hashtags in the tweet (if any).\n+   */\n+  private static class ExtractHashtagsFn extends DoFn<String, String> {\n+\n+    @Override\n+    public void processElement(ProcessContext ctx) throws Exception {\n+      for (JsonNode hashtag : jsonMapper.readTree(ctx.element())\n+          .with(\"entities\")\n+          .withArray(\"hashtags\")) {\n+        ctx.output(hashtag.get(\"text\").asText());\n+      }\n+    }\n+  }\n+\n+  // extract timestamp from \"timestamp_ms\" field.\n+  private static SerializableFunction<KV<byte[], String>, Instant> timestampFn =\n+      new SerializableFunction<KV<byte[], String>, Instant>() {\n+        @Override\n+        public Instant apply(KV<byte[], String> kv) {\n+          try {\n+            long tsMillis = jsonMapper.readTree(kv.getValue()).path(\"timestamp_ms\").asLong();\n+            return tsMillis == 0 ? Instant.now() : new Instant(tsMillis);\n+          } catch (Exception e) {\n+            throw Throwables.propagate(e);\n+          }\n+        }\n+      };\n+\n+  // return json string containing top hashtags and window information time\n+  private static class OutputFormatter extends DoFn<List<KV<String, Long>>, String>\n+      implements DoFn.RequiresWindowAccess {\n+\n+    private transient DateTimeFormatter formatter;\n+    private transient ObjectWriter jsonWriter;\n+\n+    static class OutputJson {\n+      @JsonProperty String windowStart;\n+      @JsonProperty String windowEnd;\n+      @JsonProperty String generatedAt;\n+      @JsonProperty List<HashtagInfo> topHashtags;\n+\n+      OutputJson(String windowStart, String windowEnd,\n+                 String generatedAt, List<HashtagInfo> topHashtags) {\n+        this.windowStart = windowStart;\n+        this.windowEnd = windowEnd;\n+        this.generatedAt = generatedAt;\n+        this.topHashtags = topHashtags;\n+      }\n+    }\n+\n+    static class HashtagInfo {\n+      @JsonProperty final String hashtag;\n+      @JsonProperty final long count;\n+      HashtagInfo(String hashtag, long count) {\n+        this.hashtag = hashtag;\n+        this.count = count;\n+      }\n+    }\n+\n+    @Override\n+    public void processElement(ProcessContext ctx) throws Exception {\n+      if (formatter == null) {\n+        formatter = DateTimeFormat.forPattern(\"yyyy-MM-dd HH:mm:ss\").withZoneUTC();\n+        jsonWriter = new ObjectMapper().writerWithType(OutputJson.class);\n+      }\n+\n+      List<HashtagInfo> topHashtags = new ArrayList<>(ctx.element().size());\n+\n+      for (KV<String, Long> tag : ctx.element()) {\n+        topHashtags.add(new HashtagInfo(tag.getKey(), tag.getValue()));\n+      }\n+\n+      IntervalWindow window = (IntervalWindow) ctx.window();\n+\n+      String json = jsonWriter.writeValueAsString(new OutputJson(\n+          formatter.print(window.start()),\n+          formatter.print(window.end()),\n+          formatter.print(Instant.now()),\n+          topHashtags));\n+\n+      ctx.output(json);\n+    }\n+  }\n+\n+  private static class KafkaWriter extends DoFn<String, Void> {\n+\n+    private final String topic;\n+    private final Map<String, Object> config;\n+    private transient KafkaProducer<String, String> producer = null;\n+\n+    public KafkaWriter(Options options) {\n+      this.topic = options.getOutputTopic();\n+      this.config = ImmutableMap.<String, Object>of(\n+          \"bootstrap.servers\", options.getBootstrapServers(),\n+          \"key.serializer\",    StringSerializer.class.getName(),\n+          \"value.serializer\",  StringSerializer.class.getName());\n+    }\n+\n+    @Override\n+    public void processElement(ProcessContext ctx) throws Exception {\n+      if (producer == null) {\n+        producer = new KafkaProducer<String, String>(config);","path":"contrib/examples/kafka/src/main/java/com/google/cloud/dataflow/contrib/kafka/examples/TopHashtagsExample.java","position":null,"original_position":251,"commit_id":"43973e72d38d4218d4560517a2e5e1582c7b8798","original_commit_id":"0db0a37226f775d254ff981fe4ac25f1c6e45eff","user":{"login":"rangadi","id":502522,"avatar_url":"https://avatars.githubusercontent.com/u/502522?v=3","gravatar_id":"","url":"https://api.github.com/users/rangadi","html_url":"https://github.com/rangadi","followers_url":"https://api.github.com/users/rangadi/followers","following_url":"https://api.github.com/users/rangadi/following{/other_user}","gists_url":"https://api.github.com/users/rangadi/gists{/gist_id}","starred_url":"https://api.github.com/users/rangadi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rangadi/subscriptions","organizations_url":"https://api.github.com/users/rangadi/orgs","repos_url":"https://api.github.com/users/rangadi/repos","events_url":"https://api.github.com/users/rangadi/events{/privacy}","received_events_url":"https://api.github.com/users/rangadi/received_events","type":"User","site_admin":false},"body":"done. and should I close the producer in finishBundle()?","created_at":"2016-03-18T22:41:35Z","updated_at":"2016-03-18T22:41:35Z","html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121#discussion_r56732321","pull_request_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121","_links":{"self":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments/56732321"},"html":{"href":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121#discussion_r56732321"},"pull_request":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121"}}},"pull_request":{"url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121","id":59970870,"html_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121","diff_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121.diff","patch_url":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121.patch","issue_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/121","number":121,"state":"open","locked":false,"title":"[BEAM-52] Kafka custom source ","user":{"login":"rangadi","id":502522,"avatar_url":"https://avatars.githubusercontent.com/u/502522?v=3","gravatar_id":"","url":"https://api.github.com/users/rangadi","html_url":"https://github.com/rangadi","followers_url":"https://api.github.com/users/rangadi/followers","following_url":"https://api.github.com/users/rangadi/following{/other_user}","gists_url":"https://api.github.com/users/rangadi/gists{/gist_id}","starred_url":"https://api.github.com/users/rangadi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rangadi/subscriptions","organizations_url":"https://api.github.com/users/rangadi/orgs","repos_url":"https://api.github.com/users/rangadi/repos","events_url":"https://api.github.com/users/rangadi/events{/privacy}","received_events_url":"https://api.github.com/users/rangadi/received_events","type":"User","site_admin":false},"body":"Current status (3/18/16) : well tested and feature complete. going through reviews.\r\n\r\n--\r\nThis is a preliminary PR. It is not tested yet. I am working on example application and unit tests. Some TODOs : \r\n\r\n- [x] unit tests\r\n- [ ] testing in the cluster with larger volumes of data and larger number of partitions\r\n- [x] add more stats (I need to look into stats support in sdk)\r\n","created_at":"2016-02-19T16:19:34Z","updated_at":"2016-03-18T22:41:35Z","closed_at":null,"merged_at":null,"merge_commit_sha":"1802f0444d9f4e5c75b64878b189f8ca371deb8b","assignee":{"login":"dhalperi","id":526415,"avatar_url":"https://avatars.githubusercontent.com/u/526415?v=3","gravatar_id":"","url":"https://api.github.com/users/dhalperi","html_url":"https://github.com/dhalperi","followers_url":"https://api.github.com/users/dhalperi/followers","following_url":"https://api.github.com/users/dhalperi/following{/other_user}","gists_url":"https://api.github.com/users/dhalperi/gists{/gist_id}","starred_url":"https://api.github.com/users/dhalperi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dhalperi/subscriptions","organizations_url":"https://api.github.com/users/dhalperi/orgs","repos_url":"https://api.github.com/users/dhalperi/repos","events_url":"https://api.github.com/users/dhalperi/events{/privacy}","received_events_url":"https://api.github.com/users/dhalperi/received_events","type":"User","site_admin":false},"milestone":null,"commits_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121/commits","review_comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121/comments","review_comment_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments{/number}","comments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/121/comments","statuses_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/statuses/43973e72d38d4218d4560517a2e5e1582c7b8798","head":{"label":"rangadi:kafka","ref":"kafka","sha":"43973e72d38d4218d4560517a2e5e1582c7b8798","user":{"login":"rangadi","id":502522,"avatar_url":"https://avatars.githubusercontent.com/u/502522?v=3","gravatar_id":"","url":"https://api.github.com/users/rangadi","html_url":"https://github.com/rangadi","followers_url":"https://api.github.com/users/rangadi/followers","following_url":"https://api.github.com/users/rangadi/following{/other_user}","gists_url":"https://api.github.com/users/rangadi/gists{/gist_id}","starred_url":"https://api.github.com/users/rangadi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rangadi/subscriptions","organizations_url":"https://api.github.com/users/rangadi/orgs","repos_url":"https://api.github.com/users/rangadi/repos","events_url":"https://api.github.com/users/rangadi/events{/privacy}","received_events_url":"https://api.github.com/users/rangadi/received_events","type":"User","site_admin":false},"repo":{"id":50684933,"name":"DataflowJavaSDK","full_name":"rangadi/DataflowJavaSDK","owner":{"login":"rangadi","id":502522,"avatar_url":"https://avatars.githubusercontent.com/u/502522?v=3","gravatar_id":"","url":"https://api.github.com/users/rangadi","html_url":"https://github.com/rangadi","followers_url":"https://api.github.com/users/rangadi/followers","following_url":"https://api.github.com/users/rangadi/following{/other_user}","gists_url":"https://api.github.com/users/rangadi/gists{/gist_id}","starred_url":"https://api.github.com/users/rangadi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rangadi/subscriptions","organizations_url":"https://api.github.com/users/rangadi/orgs","repos_url":"https://api.github.com/users/rangadi/repos","events_url":"https://api.github.com/users/rangadi/events{/privacy}","received_events_url":"https://api.github.com/users/rangadi/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/rangadi/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/rangadi/DataflowJavaSDK","forks_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/forks","keys_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/keys{/key_id}","collaborators_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/teams","hooks_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/hooks","issue_events_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/issues/events{/number}","events_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/events","assignees_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/assignees{/user}","branches_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/branches{/branch}","tags_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/tags","blobs_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/git/refs{/sha}","trees_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/git/trees{/sha}","statuses_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/statuses/{sha}","languages_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/languages","stargazers_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/stargazers","contributors_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/contributors","subscribers_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/subscribers","subscription_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/subscription","commits_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/commits{/sha}","git_commits_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/git/commits{/sha}","comments_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/comments{/number}","issue_comment_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/issues/comments{/number}","contents_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/contents/{+path}","compare_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/compare/{base}...{head}","merges_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/merges","archive_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/downloads","issues_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/issues{/number}","pulls_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/pulls{/number}","milestones_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/milestones{/number}","notifications_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/labels{/name}","releases_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/releases{/id}","deployments_url":"https://api.github.com/repos/rangadi/DataflowJavaSDK/deployments","created_at":"2016-01-29T19:12:02Z","updated_at":"2016-01-29T19:12:06Z","pushed_at":"2016-03-18T18:44:45Z","git_url":"git://github.com/rangadi/DataflowJavaSDK.git","ssh_url":"git@github.com:rangadi/DataflowJavaSDK.git","clone_url":"https://github.com/rangadi/DataflowJavaSDK.git","svn_url":"https://github.com/rangadi/DataflowJavaSDK","homepage":"http://cloud.google.com/dataflow","size":10890,"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":"29292f602b0060d09180bc1ff1311246de63c43c","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}","deployments_url":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/deployments","created_at":"2014-12-03T19:20:21Z","updated_at":"2016-03-18T06:37:38Z","pushed_at":"2016-03-18T18:44:46Z","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":11112,"stargazers_count":580,"watchers_count":580,"language":"Java","has_issues":true,"has_downloads":false,"has_wiki":false,"has_pages":false,"forks_count":189,"mirror_url":null,"open_issues_count":20,"forks":189,"open_issues":20,"watchers":580,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121"},"html":{"href":"https://github.com/GoogleCloudPlatform/DataflowJavaSDK/pull/121"},"issue":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/121"},"comments":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/issues/121/comments"},"review_comments":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121/comments"},"review_comment":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/pulls/121/commits"},"statuses":{"href":"https://api.github.com/repos/GoogleCloudPlatform/DataflowJavaSDK/statuses/43973e72d38d4218d4560517a2e5e1582c7b8798"}}}}