确定 PCollection 的大小

Question

我正在编写一个数据流作业，它将从 GCS 和 BigQuery 读取数据。此作业将整合从两个来源读取的数据。合并数据只是字符串。

然后此作业会将合并数据发布到外部 api.Custom 写入接收器以发布合并数据。

外部API如果合并数据超过1GB，将不允许发布数据。

如果合并数据超过 1 GB，我只想使数据流作业失败。如何获取 PCollection 中存在的数据大小？

目前我正在使用以下代码确定尺寸

    private static class CalculateSize extends PTransform<PCollection<String>, PCollection<Long>> {
    private static final long serialVersionUID = -7383871712471335638L;

    @Override
    public PCollection<Long> apply(PCollection<String> input) {
        return input
            .apply(ParDo.named("IndividualSize").of(new DoFn<String, Long>() {

                @Override
                public void processElement(ProcessContext c) throws Exception {
                    c.output(Integer.valueOf(c.element().length()).longValue());
                }
            }))
            .apply(Combine.globally(new Sum.SumLongFn()));
    }

}

还有其他更好的方法可以找到尺寸吗？

Answer 1

您发布的代码是执行此操作的正确方法。确定以预期格式写入接收器时大约会占用多少数据完全是特定于接收器的，Dataflow 无法为您执行此操作。所以，写一个函数来手动计算这是最好的方法。

请注意，您需要考虑不同来源的间接费用。例如。例如，如果您的接收器是一个 CSV 文件，那么只需将各个记录字段的长度相加，您就会低估该文件将占用的字节数。您需要考虑逗号、空格、换行符、引号、多字节字符等。此开销也完全是特定于格式的。

但是如果确保不超过 1GB 很重要，您可以简单地稍微悲观地扩大您的近似值。

确定 PCollection 的大小

Determining size of PCollection

google-cloud-dataflow