examples/dataflow-elasticsearch-indexer/ElasticIndexer/src/main/java/com/google/cloud/pso/utils/PublishProducts.java
/**
   * Runs the pipeline with the supplied options
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
Runs the pipeline with the supplied options
@param options The execution parameters to the pipeline.
@return The result of the pipeline execution.
Runs the pipeline with the supplied options @param options The execution parameters to the pipeline . @return The result of the pipeline execution .
private static PipelineResult run(PublishProductsOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    pipeline.getCoderRegistry().registerCoderForClass(JsonNode.class, JsonNodeCoder.of());
    /**
     * Steps: 1) Create a filepattern based on the input GCS path. 2) Match the file pattern and
     * read the matched patterns. 3) Read files, one per {@link ParDo} as the json messages are not
     * line delimited. The {@link DoFn} implementation generates a separate json message for every
     * product. 4) Read the product jsons and generate {@link Mutation} using {@link
     * JsonToMutationFn} 5) Write to Bigtable 6) Publish to a PubSub topic.
     */
    PCollection<JsonNode> jsonNodes =
        pipeline
            .apply("Create Filepattern", Create.of(options.getInput()))
            .apply(FileIO.matchAll().withEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW))
            .apply(FileIO.readMatches().withCompression(Compression.AUTO))
            .apply(
                "Parse and Extract",
                ParDo.of(
                    new DoFn<FileIO.ReadableFile, JsonNode>() {
                      @ProcessElement
                      public void processElement(ProcessContext context) throws IOException {
                        FileIO.ReadableFile readableFile = context.element();
                        String raw = readableFile.readFullyAsUTF8String();
                        JSONArray jsonArray = new JSONArray(raw);
                        final ObjectMapper mapper = new ObjectMapper();
                        jsonArray.forEach(
                            obj -> {
                              JSONObject jsonObject = (JSONObject) obj;
                              try {
                                context.output(mapper.readTree(jsonObject.toString()));
                              } catch (IOException e) {
                                throw new RuntimeException(e);
                              }
                            });
                      }
                    }));
    jsonNodes
        .apply(
            "Create Mutation",
            ParDo.of(
                new JsonNodeToMutationFn(
                    options.getColumnFamily(), options.getColumnQualifier(), options.getIdField())))
        .apply(
            "Write to Bigtable",
            BigtableIO.write()
                .withProjectId(options.getProject())
                .withInstanceId(options.getInstanceId())
                .withTableId(options.getTableName()));
    jsonNodes
        .apply(
            "Convert JsonNode to Strings",
            ParDo.of(
                new DoFn<JsonNode, String>() {
                  private final ObjectMapper MAPPER = new ObjectMapper();
                  @ProcessElement
                  public void processElement(ProcessContext context)
                      throws JsonProcessingException {
                    context.output(
                        MAPPER
                            .writerWithDefaultPrettyPrinter()
                            .writeValueAsString(context.element()));
                  }
                }))
        .apply("Publish to PubSub", PubsubIO.writeStrings().to(options.getTopic()));
    return pipeline.run();
  }
private static PipelineResult run ( PublishProductsOptions options ) { Pipeline pipeline = Pipeline . create ( options ) ; pipeline . getCoderRegistry ( ) . registerCoderForClass ( JsonNode . class , JsonNodeCoder . of ( ) ) ; /**
     * Steps: 1) Create a filepattern based on the input GCS path. 2) Match the file pattern and
     * read the matched patterns. 3) Read files, one per {@link ParDo} as the json messages are not
     * line delimited. The {@link DoFn} implementation generates a separate json message for every
     * product. 4) Read the product jsons and generate {@link Mutation} using {@link
     * JsonToMutationFn} 5) Write to Bigtable 6) Publish to a PubSub topic.
     */ PCollection < JsonNode > jsonNodes = pipeline . apply ( "Create Filepattern" , Create . of ( options . getInput ( ) ) ) . apply ( FileIO . matchAll ( ) . withEmptyMatchTreatment ( EmptyMatchTreatment . DISALLOW ) ) . apply ( FileIO . readMatches ( ) . withCompression ( Compression . AUTO ) ) . apply ( "Parse and Extract" , ParDo . of ( new DoFn < FileIO . ReadableFile , JsonNode > ( ) { @ ProcessElement public void processElement ( ProcessContext context ) throws IOException { FileIO . ReadableFile readableFile = context . element ( ) ; String raw = readableFile . readFullyAsUTF8String ( ) ; JSONArray jsonArray = new JSONArray ( raw ) ; final ObjectMapper mapper = new ObjectMapper ( ) ; jsonArray . forEach ( obj -> { JSONObject jsonObject = ( JSONObject ) obj ; try { context . output ( mapper . readTree ( jsonObject . toString ( ) ) ) ; } catch ( IOException e ) { throw new RuntimeException ( e ) ; } } ) ; } } ) ) ; jsonNodes . apply ( "Create Mutation" , ParDo . of ( new JsonNodeToMutationFn ( options . getColumnFamily ( ) , options . getColumnQualifier ( ) , options . getIdField ( ) ) ) ) . apply ( "Write to Bigtable" , BigtableIO . write ( ) . withProjectId ( options . getProject ( ) ) . withInstanceId ( options . getInstanceId ( ) ) . withTableId ( options . getTableName ( ) ) ) ; jsonNodes . apply ( "Convert JsonNode to Strings" , ParDo . of ( new DoFn < JsonNode , String > ( ) { private final ObjectMapper MAPPER = new ObjectMapper ( ) ; @ ProcessElement public void processElement ( ProcessContext context ) throws JsonProcessingException { context . output ( MAPPER . writerWithDefaultPrettyPrinter ( ) . writeValueAsString ( context . element ( ) ) ) ; } } ) ) . apply ( "Publish to PubSub" , PubsubIO . writeStrings ( ) . to ( options . getTopic ( ) ) ) ; return pipeline . run ( ) ; }
Runs the pipeline with the supplied options
 @param options The execution parameters to the pipeline.
Runs the pipeline with the supplied options @param options The execution parameters to the pipeline .
// Create the pipeline // Set coder for JsonNode // 1) Create a filepattern based on the input GCS path. // 2) Match the file pattern and read the //    matched patterns. // 3) Read files, one per {@link ParDo} as the json //    messages are not line delimited. //    The {@link DoFn} implementation generates a //    separate json message for every product. // 4) Read the product jsons and generate {@link Mutation} //    using {@link JsonToMutationFn} // 5) Write to Bigtable // 6) Publish to a PubSub topic.