title: Create Apache Beam Pipeline that read from Google Pub/Sub
body: <p>I am trying to create a stream pipeline using apache-beam, that read sentences from google pub/sub and write the words into a Bigquery Table.</p>
  
  <p>I am using <code>0.6.0</code> apache-beam version.</p>
  
  <p>Following the examples, I have made this:</p>
  
  <pre><code>public class StreamingWordExtract {
  
  /**
   * A DoFn that tokenizes lines of text into individual words.
   */
  static class ExtractWords extends DoFn&lt;String, String&gt; {
      @ProcessElement
      public void processElement(ProcessContext c) {
          String[] words = ((String) c.element()).split("[^a-zA-Z']+");
          for (String word : words) {
              if (!word.isEmpty()) {
                  c.output(word);
              }
          }
      }
  }
  
  /**
   * A DoFn that uppercases a word.
   */
  static class Uppercase extends DoFn&lt;String, String&gt; {
      @ProcessElement
      public void processElement(ProcessContext c) {
          c.output(c.element().toUpperCase());
      }
  }
  
  
  /**
   * A DoFn that uppercases a word.
   */
  static class StringToRowConverter extends DoFn&lt;String, TableRow&gt; {
      @ProcessElement
      public void processElement(ProcessContext c) {
          c.output(new TableRow().set("string_field", c.element()));
      }
  
      static TableSchema getSchema() {
          return new TableSchema().setFields(new ArrayList&lt;TableFieldSchema&gt;() {
              // Compose the list of TableFieldSchema from tableSchema.
              {
                  add(new TableFieldSchema().setName("string_field").setType("STRING"));
              }
          });
      }
  
  }
  
  private interface StreamingWordExtractOptions extends ExampleBigQueryTableOptions, ExamplePubsubTopicOptions {
      @Description("Input file to inject to Pub/Sub topic")
      @Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
      String getInputFile();
  
      void setInputFile(String value);
  }
  
  public static void main(String[] args) {
      StreamingWordExtractOptions options = PipelineOptionsFactory.fromArgs(args)
              .withValidation()
              .as(StreamingWordExtractOptions.class);
  
      options.setBigQuerySchema(StringToRowConverter.getSchema());
  
      Pipeline p = Pipeline.create(options);
  
      String tableSpec = new StringBuilder()
              .append(options.getProject()).append(":")
              .append(options.getBigQueryDataset()).append(".")
              .append(options.getBigQueryTable())
              .toString();
  
      p.apply(PubsubIO.read().topic(options.getPubsubTopic()))
              .apply(ParDo.of(new ExtractWords()))
              .apply(ParDo.of(new StringToRowConverter()))
              .apply(BigQueryIO.Write.to(tableSpec)
                      .withSchema(StringToRowConverter.getSchema())
                      .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                      .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
  
      PipelineResult result = p.run();
  
  
  }
  </code></pre>
  
  <p>I have an error near:</p>
  
  <p><code>apply(ParDo.of(new ExtractWords()))</code></p>
  
  <p>because the previous <code>apply</code> not return a <code>String</code> but an <code>Object</code></p>
  
  <p>I suppose that the problem is the type returned from <code>PubsubIO.read().topic(options.getPubsubTopic())</code>. The type is <code>PTransform&lt;PBegin, PCollection&lt;T&gt;&gt;</code> instead of <code>PTransform&lt;PBegin, PCollection&lt;String&gt;&gt;</code></p>
  
  <p>Which is the correct way to read from google pub/sub using apache-beam?</p>