确定PCollection的大小

Question

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

然后此作业将把合并后的数据发布到外部API中。编写自定义接收器以发布合并后的数据。

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

我只想在合并数据超过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，那么您可以稍微悲观地扩大您的近似值。