在调试器中调试apache beam/dataflow？

Question

这首先与这篇文章高度相关->如何在Dataflow中进行这种类型的测试（在twitter上称为功能测试）？

我们在生产中有一些类似的代码

@Override
public PDone expand(PCollectionTuple in) {
    final RosterPipelineOptions options = (RosterPipelineOptions) in.getPipeline().getOptions();
    try {
        final Schema schema = new Schema.Parser().parse(PractitionerStandardAvroWriter.class.getResourceAsStream("/standardFormat.avsc"));
        final String schemaStr = new BufferedReader(new InputStreamReader(PractitionerStandardAvroWriter.class.getResourceAsStream("/standardFormat.avsc")))
            .lines().collect(Collectors.joining("\n"));

        final PCollection<Validated<PractitionerStandardOutputDto>> validOutputs = in.get(PractitionerStandardOutputConverter.VALID_OUTPUTS_TAG);
        final PCollection<Validated<PractitionerStandardOutputDto>> invalidOutputs = in.get(PractitionerStandardOutputConverter.INVALID_OUTPUTS_TAG);

        final PCollection<GenericRecord> validRecords = validOutputs.apply(
            "Transform Valid Standard Output into Generic Rows", ParDo.of(new GenericRecordConverter(schemaStr)));
        final PCollection<GenericRecord> invalidRecords = invalidOutputs.apply(
            "Transform Invalid Standard Output into Generic Rows", ParDo.of(new GenericRecordConverter(schemaStr)));

        validRecords
            .setCoder(AvroCoder.of(GenericRecord.class, schema))
            .apply("Write Records", AvroIO.writeGenericRecords(schema)
                .to(options.getValidOutput())
                .withoutSharding());

        final PCollection<String> invalidRows = invalidRecords
            .setCoder(AvroCoder.of(GenericRecord.class, schema))
            .apply("Convert Error Avro to Csv", ParDo.of(new AvroToCsvConvertFn(schemaStr, ",")));
        invalidRows.apply("Write Error Records to Csv",
            TextIO.write().to(options.getInvalidOutput()).withoutSharding());

        return PDone.in(in.getPipeline());
    }
    catch (IOException e) {
        SneakyThrow.sneak(e); return null;
    }
}

• 这是编写可调试的Apache-Beam/Dataflow代码的最佳方式吗？我们可以逐步通过这些代码并轻松地看到我们的bug在哪里？
• 是否有其他方法可以方便地调试它，因为我怀疑在应用stuff时，“真正的执行”发生在该方法之后？

谢谢，迪恩

Answer 1

总的来说，就像你的另一个问题一样，我的建议是：

1. 要单步执行管道，可以编写一个用IDE运行的单元测试，它将在Directrunner中运行。这使您可以轻松地单步执行您的管道。这不是在数据流中运行的，而是在本地运行的--它仍然很有价值。

您可以在expand中设置断点，这些断点将在管道构建时命中。您可以在DoFn的进程中设置断点，或者在源代码的拆分、读取中设置断点--这些断点将在流水线执行时命中。

一些最有价值的测试工具是passert、TestStream和TestPipeline。我建议您查看我共享的页面，看看这些实用程序是否会有所帮助。

对于特定的管道，我可能认为可以将PTransform分成更小的部分，并为每个部分编写简单的单元测试。