Add support for deduplicating id_label in PubSubSource

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=117988854

Author: charlesccychen

google/cloud/dataflow/io/pubsub.py

@@ -24,11 +24,26 @@
 
 
 class PubSubSource(iobase.Source):
-  """Source for reading from a given Cloud Pub/Sub topic."""
+  """Source for reading from a given Cloud Pub/Sub topic.
 
-  def __init__(self, topic, subscription=None, coder=coders.StrUtf8Coder()):
+  Attributes:
+    topic: Cloud Pub/Sub topic in the form "/topics/<project>/<topic>".
+    subscription: Optional existing Cloud Pub/Sub subscription to use in the
+      form "projects/<project>/subscriptions/<subscription>".
+    id_label: The attribute on incoming Pub/Sub messages to use as a unique
+      record identifier.  When specified, the value of this attribute (which can
+      be any string that uniquely identifies the record) will be used for
+      deduplication of messages.  If not provided, Dataflow cannot guarantee
+      that no duplicate data will be delivered on the Pub/Sub stream. In this
+      case, deduplication of the stream will be strictly best effort.
+    coder: The Coder to use for decoding incoming Pub/Sub messages.
+  """
+
+  def __init__(self, topic, subscription=None, id_label=None,
+               coder=coders.StrUtf8Coder()):
     self.topic = topic
     self.subscription = subscription
+    self.id_label = id_label
     self.coder = coder
 
   @property

google/cloud/dataflow/runners/dataflow_runner.py

@@ -488,6 +488,9 @@ def run_Read(self, transform_node):
       if transform.source.subscription:
         step.add_property(PropertyNames.PUBSUB_SUBSCRIPTION,
                           transform.source.topic)
+      if transform.source.id_label:
+        step.add_property(PropertyNames.PUBSUB_ID_LABEL,
+                          transform.source.id_label)
     elif transform.source.format == 'custom':
       # TODO(silviuc): Implement custom sources.
       raise NotImplementedError

google/cloud/dataflow/utils/names.py

@@ -58,6 +58,7 @@ class PropertyNames(object):
   PARALLEL_INPUT = 'parallel_input'
   PUBSUB_TOPIC = 'pubsub_topic'
   PUBSUB_SUBSCRIPTION = 'pubsub_subscription'
+  PUBSUB_ID_LABEL = 'pubsub_id_label'
   SERIALIZED_FN = 'serialized_fn'
   SHARD_NAME_TEMPLATE = 'shard_template'
   STEP_NAME = 'step_name'