examples/dataflow-elasticsearch-indexer/ElasticIndexer/src/main/java/com/google/cloud/pso/IndexerMain.java
/**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
Runs the pipeline to completion with the specified options.
@param options The execution options.
@return The pipeline result.
Runs the pipeline to completion with the specified options . @param options The execution options . @return The pipeline result .
private static PipelineResult run(IndexerOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForClass(ErrorMessage.class, ErrorMessageCoder.of());
    String[] addresses =
        Lists.newArrayList(Splitter.on(",").trimResults().split(options.getAddresses()))
            .toArray(new String[0]);
    ElasticsearchIO.ConnectionConfiguration connection =
        ElasticsearchIO.ConnectionConfiguration.create(
            addresses, options.getIndex(), options.getType());
    /**
     * The pipeline executes the following steps: 1. Read messages as Strings from PubSub. 2.
     * Validate messages and tag them with success and failure tags. 3. Extract metadata for
     * successfully validated messages. 4. Apply metadata extract in the previous step. 5.
     * Successful messages are written to Elasticsearch index. 6. Failed messages (wrapped as
     * ErrorMessage objects) are written to a PubSub topic for rejected messages.
     */
    PCollectionTuple validated =
        pipeline
            .apply(
                "Read Messages from PubSub",
                PubsubIO.readStrings().fromSubscription(options.getInputSubscription()))
            .apply(
                Window.<String>into(new GlobalWindows())
                    .triggering(AfterPane.elementCountAtLeast(1))
                    .discardingFiredPanes())
            .apply(
                "Validate Messages",
                FailSafeValidate.newBuilder()
                    .withSuccessTag(VALIDATION_SUCCESS_TAG)
                    .withFailureTag(FAILURE_TAG)
                    .withKeyPath(options.getIdField())
                    .build());
    PCollectionTuple metadataApplied =
        validated
            .get(VALIDATION_SUCCESS_TAG)
            .apply(
                "Extract Metadata",
                ExtractMetadata.newBuilder()
                    .withBigtableInstanceId(options.getInstanceId())
                    .withBigtableProjectId(options.getProject())
                    .withBigtableTableName(options.getTableName())
                    .withColumnFamily(options.getColumnFamily())
                    .withColumnQualifier(options.getColumnQualifier())
                    .withSuccessTag(METADATA_SUCCESS_TAG)
                    .withFailureTag(FAILURE_TAG)
                    .build());
    metadataApplied
        .get(METADATA_SUCCESS_TAG)
        .apply("Insert Metadata", ParDo.of(new InsertMetadataFn(options.getColumnQualifier())))
        .apply(
            "Write to Elasticsearch Index",
            ElasticsearchIO.write()
                .withConnectionConfiguration(connection)
                .withIdFn(new ExtractKeyFn(options.getIdField()))
                .withUsePartialUpdate(true));
    PCollectionList.of(validated.get(FAILURE_TAG))
        .and(metadataApplied.get(FAILURE_TAG))
        .apply("Flatten Failed Messages", Flatten.pCollections())
        .apply(
            "Write Failed Messages",
            PubsubIO.writeAvros(ErrorMessage.class).to(options.getRejectionTopic()));
    return pipeline.run();
  }
private static PipelineResult run ( IndexerOptions options ) { Pipeline pipeline = Pipeline . create ( options ) ; CoderRegistry coderRegistry = pipeline . getCoderRegistry ( ) ; coderRegistry . registerCoderForClass ( ErrorMessage . class , ErrorMessageCoder . of ( ) ) ; String [ ] addresses = Lists . newArrayList ( Splitter . on ( "," ) . trimResults ( ) . split ( options . getAddresses ( ) ) ) . toArray ( new String [ 0 ] ) ; ElasticsearchIO . ConnectionConfiguration connection = ElasticsearchIO . ConnectionConfiguration . create ( addresses , options . getIndex ( ) , options . getType ( ) ) ; /**
     * The pipeline executes the following steps: 1. Read messages as Strings from PubSub. 2.
     * Validate messages and tag them with success and failure tags. 3. Extract metadata for
     * successfully validated messages. 4. Apply metadata extract in the previous step. 5.
     * Successful messages are written to Elasticsearch index. 6. Failed messages (wrapped as
     * ErrorMessage objects) are written to a PubSub topic for rejected messages.
     */ PCollectionTuple validated = pipeline . apply ( "Read Messages from PubSub" , PubsubIO . readStrings ( ) . fromSubscription ( options . getInputSubscription ( ) ) ) . apply ( Window . < String > into ( new GlobalWindows ( ) ) . triggering ( AfterPane . elementCountAtLeast ( 1 ) ) . discardingFiredPanes ( ) ) . apply ( "Validate Messages" , FailSafeValidate . newBuilder ( ) . withSuccessTag ( VALIDATION_SUCCESS_TAG ) . withFailureTag ( FAILURE_TAG ) . withKeyPath ( options . getIdField ( ) ) . build ( ) ) ; PCollectionTuple metadataApplied = validated . get ( VALIDATION_SUCCESS_TAG ) . apply ( "Extract Metadata" , ExtractMetadata . newBuilder ( ) . withBigtableInstanceId ( options . getInstanceId ( ) ) . withBigtableProjectId ( options . getProject ( ) ) . withBigtableTableName ( options . getTableName ( ) ) . withColumnFamily ( options . getColumnFamily ( ) ) . withColumnQualifier ( options . getColumnQualifier ( ) ) . withSuccessTag ( METADATA_SUCCESS_TAG ) . withFailureTag ( FAILURE_TAG ) . build ( ) ) ; metadataApplied . get ( METADATA_SUCCESS_TAG ) . apply ( "Insert Metadata" , ParDo . of ( new InsertMetadataFn ( options . getColumnQualifier ( ) ) ) ) . apply ( "Write to Elasticsearch Index" , ElasticsearchIO . write ( ) . withConnectionConfiguration ( connection ) . withIdFn ( new ExtractKeyFn ( options . getIdField ( ) ) ) . withUsePartialUpdate ( true ) ) ; PCollectionList . of ( validated . get ( FAILURE_TAG ) ) . and ( metadataApplied . get ( FAILURE_TAG ) ) . apply ( "Flatten Failed Messages" , Flatten . pCollections ( ) ) . apply ( "Write Failed Messages" , PubsubIO . writeAvros ( ErrorMessage . class ) . to ( options . getRejectionTopic ( ) ) ) ; return pipeline . run ( ) ; }
Runs the pipeline to completion with the specified options.
Runs the pipeline to completion with the specified options .
// Set the CustomCoder for the ErrorMessage class. // 1. Read messages as Strings from PubSub. // Note that streaming pipelines have GlobalWindows by default. // This explicit specification is meant to indicate that more // advanced windowing/triggering can be added to the pipeline as // needed. // 2. Validate messages and tag them with success and failure tags. // 3. Extract metadata for successfully validated messages. // 4. Apply metadata extract in the previous step. // 5. Successful messages are written to Elasticsearch index. // 6. Failed messages (wrapped as ErrorMessage objects) are written // to a PubSub topic for rejected messages.