Forbid use of PubSub I/O in batch and local jobs

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

Author: charlesccychen

google/cloud/dataflow/io/pubsub.py

@@ -36,6 +36,10 @@ def format(self):
     """Source format name required for remote execution."""
     return 'pubsub'
 
+  def reader(self):
+    raise NotImplementedError(
+        'PubSubSource is not supported in local execution.')
+
 
 class PubSubSink(iobase.NativeSink):
   """Sink for writing to a given Cloud Pub/Sub topic."""
@@ -48,3 +52,7 @@ def __init__(self, topic, coder=coders.StrUtf8Coder()):
   def format(self):
     """Sink format name required for remote execution."""
     return 'pubsub'
+
+  def writer(self):
+    raise NotImplementedError(
+        'PubSubSink is not supported in local execution.')

google/cloud/dataflow/runners/dataflow_runner.py

@@ -460,6 +460,11 @@ def run_Read(self, transform_node):
                          ' a query',
                          transform.source)
     elif transform.source.format == 'pubsub':
+      standard_options = (
+          transform_node.inputs[0].pipeline.options.view_as(StandardOptions))
+      if not standard_options.streaming:
+        raise ValueError('PubSubSource is currently only available for use in '
+                         'streaming pipelines.')
       step.add_property(PropertyNames.PUBSUB_TOPIC, transform.source.topic)
       if transform.source.subscription:
         step.add_property(PropertyNames.PUBSUB_SUBSCRIPTION,
@@ -526,6 +531,11 @@ def run__NativeWrite(self, transform_node):
         step.add_property(
             PropertyNames.BIGQUERY_SCHEMA, transform.sink.schema_as_json())
     elif transform.sink.format == 'pubsub':
+      standard_options = (
+          transform_node.inputs[0].pipeline.options.view_as(StandardOptions))
+      if not standard_options.streaming:
+        raise ValueError('PubSubSink is currently only available for use in '
+                         'streaming pipelines.')
       step.add_property(PropertyNames.PUBSUB_TOPIC, transform.sink.topic)
     else:
       raise ValueError(