@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 debugBatchPipelineWithContextfulTransform() {
    PipelineOptions options = contextRule.configure(PipelineOptionsFactory.create());
    options.setRunner(SparkRunnerDebugger.class);
    Pipeline pipeline = Pipeline.create(options);

    final PCollectionView<Integer> view =
        pipeline.apply("Dummy", Create.of(0)).apply(View.asSingleton());

    pipeline
        .apply(Create.of(Arrays.asList(0)))
        .setCoder(VarIntCoder.of())
        .apply(
            MapElements.into(new TypeDescriptor<Integer>() {})
                .via(fn((element, c) -> element, requiresSideInputs(view))));

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

    final String expectedPipeline =
        "sparkContext.<impulse>()\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Create$Values$2())\n"
            + "_.aggregate(..., new org.apache.beam.sdk.transforms.View$SingletonCombineFn(), ...)\n"
            + "_.<createPCollectionView>\n"
            + "sparkContext.<impulse>()\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Create$Values$2())\n"
            + "_.mapPartitions(new org.apache.beam.sdk.transforms.Contextful())";

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