title: Apache Beam - Python - Streaming to BigQuery writes no data to the table
body: <h1>Apache Beam - Python - Streaming to BigQuery writes no data to the table</h1>
  
  <p>I have designed a simple Apache Beam Pipeline using the Python SDK, while I know that the streaming capabilities of the Python SDK are still being developed I have stumbled upon a roadblock I cannot seem to circumvent: everything in the Pipeline works fine, until the point where I try to stream into a BigQuery table. I get no error, exception or warning, the data simply does not appear on BigQuery.</p>
  
  <ul>
  <li><p>I tried both with 2.1.0 and 2.2.0 and it yields the same results.</p></li>
  <li><p>The PubSub topic where I am fetching the data from simply contains tweets like "I love Apache Beam #apachebeam #dataflow #beam #datascience".</p></li>
  </ul>
  
  <p>This is the Pipeline:</p>
  
  <pre><code>import datetime
  import logging
  
  import apache_beam as beam
  import apache_beam.transforms.window as window
  
  from apache_beam.options.pipeline_options import GoogleCloudOptions
  from apache_beam.options.pipeline_options import PipelineOptions
  from apache_beam.options.pipeline_options import SetupOptions
  from apache_beam.options.pipeline_options import StandardOptions
  
  # Configuration
  # I set all the constant before creating the Pipeline Options
  options = PipelineOptions()
  options.view_as(StandardOptions).streaming = True
  options.view_as(StandardOptions).runner = 'DirectRunner'
  # options.view_as(SetupOptions).save_main_session = True
  # options.view_as(StandardOptions).runner = 'DataflowRunner'
  google_cloud_options = options.view_as(GoogleCloudOptions)
  google_cloud_options.project = PROJECT_ID
  
  
  # DoFn
  class ExtractHashtagsDoFn(beam.DoFn):
      def process(self, element):
          """
          Takes a tweet as input, search and return its content 
          and hashtag as a dictionary.
          """
          hastags = [w for w in element.split() if w.startswith('#')]
          result = {'tweet': element, 'hashtags': hastags}
          logging.info('{}'.format(result))
          yield result
  
  
  class SentimentAnalysisDoFn(beam.DoFn):
      def process(self, element):
          from google.cloud import language
          from google.gax.errors import RetryError
  
          client = language.LanguageServiceClient()
          document = language.types.Document(
              content=element['tweet'].encode('utf-8'),
              type='PLAIN_TEXT')
  
          try:
              response = client.analyze_sentiment(
                  document=document,
                  encoding_type='UTF8')
              sentiment = response.document_sentiment
              element['sentiment_score'] = sentiment.score
              element['sentiment_magnitude'] = sentiment.magnitude
              element['analyzed'] = 1
  
          except RetryError:
              element['sentiment_score'] = None
              element['sentiment_magnitude'] = None
              element['analyzed'] = 0
  
  #         logging.info('{}'.format(element))
          yield element
  
  
  # Define Composite Transforms
  class TextAnalysisTransform(beam.PTransform):
      def expand(self, pcoll):
          return(
              pcoll
              | 'Decode' &gt;&gt; beam.Map(lambda string: string.decode('utf8', 'ignore'))
              | 'ExtractHashtags' &gt;&gt; beam.ParDo(ExtractHashtagsDoFn())
              | 'SentimentAnalysis' &gt;&gt; beam.ParDo(SentimentAnalysisDoFn())
          )
  
  
  class WindowingForOutputTransform(beam.PTransform):
      def expand(self, pcoll):
          import json
          return(
              pcoll
              | 'Pack' &gt;&gt; beam.Map(lambda x: (x, 1))
              | 'Windowing' &gt;&gt; beam.WindowInto(window.FixedWindows(5, 0))
              | 'GroupByKey' &gt;&gt; beam.GroupByKey()
              | 'Unpack' &gt;&gt; beam.Map(lambda x: x[0])
          )
  
  
  class SaveToBigQueryTransform(beam.PTransform):
      def expand(self, pcoll):
          # Define the Schema.
          from apache_beam.io.gcp.internal.clients import bigquery
  
          table_schema = bigquery.TableSchema()
  
          # Fields that use standard types.
          alpha_schema = bigquery.TableFieldSchema()
          alpha_schema.name = 'tweet'
          alpha_schema.type = 'string'
          alpha_schema.mode = 'nullable'
          table_schema.fields.append(alpha_schema)
  
          beta_schema = bigquery.TableFieldSchema()
          beta_schema.name = 'sentiment_score'
          beta_schema.type = 'float'
          beta_schema.mode = 'nullable'
          table_schema.fields.append(beta_schema)
  
          gamma_schema = bigquery.TableFieldSchema()
          gamma_schema.name = 'sentiment_magnitude'
          gamma_schema.type = 'float'
          gamma_schema.mode = 'nullable'
          table_schema.fields.append(gamma_schema)
  
          delta_schema = bigquery.TableFieldSchema()
          delta_schema.name = 'analyzed'
          delta_schema.type = 'integer'
          delta_schema.mode = 'required'
          table_schema.fields.append(delta_schema)
  
          # A repeated field.
          omega_schema = bigquery.TableFieldSchema()
          omega_schema.name = 'hashtags'
          omega_schema.type = 'string'
          omega_schema.mode = 'repeated'
          table_schema.fields.append(omega_schema)
  
          # Saving the output to BigQuery.
          return (
              pcoll
              | 'PrepareForOutput' &gt;&gt; WindowingForOutputTransform()
              | 'WriteToBigQuery' &gt;&gt; beam.io.WriteToBigQuery(
                  table='twitter_analysis',
                  dataset=DATASET_ID,
                  project=PROJECT_ID,
                  schema=table_schema,  # Pass the defined table_schema
                  create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                  write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
  
  
  def run(argv=None):
      """
      Define and run the pipeline.
      """
      with beam.Pipeline(options=options) as p:
          plumbing = (
              p
              | 'LoadTestData' &gt;&gt; beam.io.ReadStringsFromPubSub(subscription=FULL_SUBSCRIPTION)
              | 'TextAnalysis' &gt;&gt; TextAnalysisTransform()
              | 'StreamToBigQuery' &gt;&gt; SaveToBigQueryTransform()
          )
  
  if __name__ == '__main__':
      logging.getLogger().setLevel(logging.INFO)
      run()
  </code></pre>
  
  <blockquote>
    <p><strong>Output:</strong>
    /home/ubik/Documents/github/twitter-streaming/beam/.venv/lib/python2.7/site-packages/apache_beam/coders/typecoders.py:135:
    UserWarning: Using fallback coder for typehint: Any.
              warnings.warn('Using fallback coder for typehint: %r.' % typehint)
            INFO:root:Running pipeline with DirectRunner.</p>
  </blockquote>
  
  <p><strong>EDIT:</strong> I tried to run the pipeline again on September 4, everything worked fine. Probably the BigQuery poltergeist stopped haunting me.</p>