@Override
  public SparkPipelineResult run(Pipeline pipeline) {
    boolean isStreaming =
        options.isStreaming() || options.as(TestSparkPipelineOptions.class).isForceStreaming();

    // Default to using the primitive versions of Read.Bounded and Read.Unbounded.
    // TODO(https://github.com/apache/beam/issues/20530): Use SDF read as default when we address
    // performance issue.
    if (!ExperimentalOptions.hasExperiment(pipeline.getOptions(), "beam_fn_api")) {
      SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(pipeline);
    }

    JavaSparkContext jsc =
        SparkContextFactory.getSparkContext(pipeline.getOptions().as(SparkPipelineOptions.class));
    JavaStreamingContext jssc =
        new JavaStreamingContext(jsc, new org.apache.spark.streaming.Duration(1000));

    SparkRunner.initAccumulators(options, jsc);

    TransformTranslator.Translator translator = new TransformTranslator.Translator();

    SparkNativePipelineVisitor visitor;
    if (isStreaming) {
      SparkPipelineTranslator streamingTranslator =
          new StreamingTransformTranslator.Translator(translator);
      EvaluationContext ctxt = new EvaluationContext(jsc, pipeline, options, jssc);
      visitor = new SparkNativePipelineVisitor(streamingTranslator, ctxt);
    } else {
      EvaluationContext ctxt = new EvaluationContext(jsc, pipeline, options, jssc);
      visitor = new SparkNativePipelineVisitor(translator, ctxt);
    }

    pipeline.traverseTopologically(visitor);

    SparkContextFactory.stopSparkContext(jsc);

    String debugString = visitor.getDebugString();
    LOG.info("Translated Native Spark pipeline:\n{}", debugString);
    return new DebugSparkPipelineResult(debugString);
  }
@Test
  public void debugBatchPipeline() {
    PipelineOptions options = contextRule.configure(PipelineOptionsFactory.create());
    options.setRunner(SparkRunnerDebugger.class);
    Pipeline pipeline = Pipeline.create(options);

    PCollection<String> lines =
        pipeline.apply(Create.of(Collections.<String>emptyList()).withCoder(StringUtf8Coder.of()));

    PCollection<KV<String, Long>> wordCounts = lines.apply(new WordCount.CountWords());

    wordCounts.apply(GroupByKey.create()).apply(Combine.groupedValues(Sum.ofLongs()));

    PCollection<KV<String, Long>> wordCountsPlusOne =
        wordCounts.apply(MapElements.via(new PlusOne()));

    PCollectionList.of(wordCounts).and(wordCountsPlusOne).apply(Flatten.pCollections());

    wordCounts
        .apply(MapElements.via(new WordCount.FormatAsTextFn()))
        .apply(TextIO.write().to("!!PLACEHOLDER-OUTPUT-DIR!!").withNumShards(3).withSuffix(".txt"));

    final String expectedPipeline =
        "sparkContext.<impulse>()\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.FlatMapElements$3())\n"
            + "_.mapPartitions("
            + "new org.apache.beam.runners.spark.examples.WordCount$ExtractWordsFn())\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Count$PerElement$1())\n"
            + "_.combineByKey(..., new org.apache.beam.sdk.transforms.Count$CountFn(), ...)\n"
            + "_.groupByKey()\n"
            + "_.map(new org.apache.beam.sdk.transforms.Sum$SumLongFn())\n"
            + "_.mapPartitions("
            + "new org.apache.beam.runners.spark.SparkRunnerDebuggerTest$PlusOne())\n"
            + "sparkContext.union(...)\n"
            + "_.mapPartitions("
            + "new org.apache.beam.runners.spark.examples.WordCount$FormatAsTextFn())\n"
            + "_.<org.apache.beam.sdk.io.TextIO$Write>";

    SparkRunnerDebugger.DebugSparkPipelineResult result =
        (SparkRunnerDebugger.DebugSparkPipelineResult) pipeline.run();

    assertThat(
        "Debug pipeline did not equal expected",
        result.getDebugString(),
        Matchers.equalTo(expectedPipeline));
  }