examples/iot-nirvana/pipeline/src/main/java/com/google/cloud/demo/iot/nirvana/pipeline/TemperaturePipeline.java
/**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
Runs the pipeline to completion with the specified options. This method does not wait until the
pipeline is finished before returning. Invoke  result.waitUntilFinish() on the result
object to block until the pipeline is finished running if blocking programmatic execution is
required.

@param options The execution options.
@return The pipeline result.
Runs the pipeline to completion with the specified options . This method does not wait until the pipeline is finished before returning . Invoke result . waitUntilFinish () on the result object to block until the pipeline is finished running if blocking programmatic execution is required . @param options The execution options . @return The pipeline result .
public static PipelineResult run(TemperaturePipelineOptions options) throws FormatException {
    Pipeline p = Pipeline.create(options);
    PCollection<String> pubSubMessages =
        p.apply("PubSubRead", PubsubIO.readStrings().fromTopic(options.getInputTopic()));
    PCollectionView<Map<String, City>> cities = loadCitiesSideInput(p);
    PCollectionTuple filteredMessages =
        pubSubMessages.apply(
            "FilterMessages",
            ParDo.of(FilterMessages.newBuilder().setCities(cities).setErrorTag(ERROR_TAG).build())
                .withSideInputs(cities)
                .withOutputTags(GOOD_TAG, TupleTagList.of(ERROR_TAG)));
    PCollection<Message> validTemperatures = filteredMessages.get(GOOD_TAG);
    PCollection<String> invalidTemperatures = filteredMessages.get(ERROR_TAG);
    validTemperatures
        .apply("MessageToTableRow", ParDo.of(new MessageToTableRow(PARIS_TIME_ZONE)))
        .apply(
            "BigQueryWrite",
            BigQueryIO.writeTableRows()
                .to(options.getTable())
                .withSchema(MessageToTableRow.getSchema())
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
    validTemperatures
        .apply("MessageToEntity", ParDo.of(new MessageToEntity()))
        .apply("DatastoreWrite", DatastoreIO.v1().write().withProjectId(options.getProject()));
    invalidTemperatures
        .apply("OneMinuteWindows", Window.into(FixedWindows.of(Duration.standardMinutes(1))))
        .apply(
            "WriteInvalidMessages",
            TextIO.write()
                .to(options.getErrLocationPrefix() + "_" + System.currentTimeMillis())
                .withNumShards(1)
                .withSuffix(".txt")
                .withWindowedWrites());
    return p.run();
  }
public static PipelineResult run ( TemperaturePipelineOptions options ) throws FormatException { Pipeline p = Pipeline . create ( options ) ; PCollection < String > pubSubMessages = p . apply ( "PubSubRead" , PubsubIO . readStrings ( ) . fromTopic ( options . getInputTopic ( ) ) ) ; PCollectionView < Map < String , City > > cities = loadCitiesSideInput ( p ) ; PCollectionTuple filteredMessages = pubSubMessages . apply ( "FilterMessages" , ParDo . of ( FilterMessages . newBuilder ( ) . setCities ( cities ) . setErrorTag ( ERROR_TAG ) . build ( ) ) . withSideInputs ( cities ) . withOutputTags ( GOOD_TAG , TupleTagList . of ( ERROR_TAG ) ) ) ; PCollection < Message > validTemperatures = filteredMessages . get ( GOOD_TAG ) ; PCollection < String > invalidTemperatures = filteredMessages . get ( ERROR_TAG ) ; validTemperatures . apply ( "MessageToTableRow" , ParDo . of ( new MessageToTableRow ( PARIS_TIME_ZONE ) ) ) . apply ( "BigQueryWrite" , BigQueryIO . writeTableRows ( ) . to ( options . getTable ( ) ) . withSchema ( MessageToTableRow . getSchema ( ) ) . withCreateDisposition ( BigQueryIO . Write . CreateDisposition . CREATE_IF_NEEDED ) . withWriteDisposition ( BigQueryIO . Write . WriteDisposition . WRITE_APPEND ) ) ; validTemperatures . apply ( "MessageToEntity" , ParDo . of ( new MessageToEntity ( ) ) ) . apply ( "DatastoreWrite" , DatastoreIO . v1 ( ) . write ( ) . withProjectId ( options . getProject ( ) ) ) ; invalidTemperatures . apply ( "OneMinuteWindows" , Window . into ( FixedWindows . of ( Duration . standardMinutes ( 1 ) ) ) ) . apply ( "WriteInvalidMessages" , TextIO . write ( ) . to ( options . getErrLocationPrefix ( ) + "_" + System . currentTimeMillis ( ) ) . withNumShards ( 1 ) . withSuffix ( ".txt" ) . withWindowedWrites ( ) ) ; return p . run ( ) ; }
Runs the pipeline to completion with the specified options.
Runs the pipeline to completion with the specified options .
// Read messages from Pub/Sub // Filter out messages that don't contain temperatures for actual cities // Convert valid messages into TableRow and write them to BigQuery // Convert valid messages to Entity objects and write them to Datastore // Write invalid messages to Google Cloud Storage in 1 minute windows // Execute pipeline