如何在GCP数据流中读取CombineFn函数的日志消息？

Question

我正在创建一个Apache Beam streaming处理管道，以在GCP Dataflow中运行。我有许多扩展DoFn和combinefn的转换。在DoFn中，使用数据流作业详细信息中的日志窗口，日志可以很好地可视化。但是，不显示来自CombineFn转换的日志。

import java.io.Serializable;
import org.apache.avro.reflect.Nullable;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class AverageSpv extends CombineFn<String, AverageSpv.Accum, String> {
    private static final Logger LOG = LoggerFactory.getLogger(AverageSpv.class);

    @DefaultCoder(AvroCoder.class)
    public static class Accum implements Serializable {
        @Nullable String id;
    }

    @Override
    public Accum createAccumulator() {
        return new Accum();
    }

    @Override
    public Accum addInput(Accum accumulator, String input) {
        LOG.info("Add input: id {}, input);

        accumulator.id = input;

        return accumulator;
    }

    @Override
    public Accum mergeAccumulators(Iterable<Accum> accumulators) {
        LOG.info("Merging accumulator");

        Accum merged = createAccumulator();
        for (Accum accumulator : accumulators) {
            merged.id = accumulator.id;            
        }

        return merged;
    }

    @Override
    public VehicleSpeedPerSegmentInfo extractOutput(Accum accumulator) {
        LOG.info("Extracting accumulator");

        LOG.info("Extract output: id {}", acummulator.id);

        return acummulator.id;
    }
}

Answer 1

Apache Beam CombineFn操作在数据流中跨几个步骤执行。（具体地说，在将所有结果洗牌到单个键之前会进行大量的预合并，然后在后续的post-GBK步骤中将所有上游结果合并到最终结果中。）事实上，图中没有一个单独的执行“步骤”对应于原始的合并步骤，这可能是阻止找到日志的原因。

这是一个bug，应该修复。如前所述，一个解决方法是查看管道中的所有日志。