与 BigQuery 表输入大小相比，为什么我的 PCollection (SCollection) 大小如此之大？答案

【问题标题】：Why is my PCollection (SCollection) size so larged compared to BigQuery Table input size?与 BigQuery 表输入大小相比，为什么我的 PCollection (SCollection) 大小如此之大？
【发布时间】：2018-05-11 23:15:23
【问题描述】：

上图是一个大查询表的表模式，它是在 spotify 的 scio 上运行的 apache Beam 数据流作业的输入。如果您不熟悉 scio，它是 Apache Beam Java SDK 的 Scala 包装器。特别是，“SCollection 包装 PCollection”。我在 BigQuery 磁盘上的输入表是 136 GB，但在数据流 UI 中查看我的 SCollection 的大小时，它是 504.91 GB。

我知道 BigQuery 在数据压缩和表示方面可能要好得多，但大小增加超过 3 倍似乎相当高。非常清楚，我使用的是类型安全大查询案例类（我们称之为 Clazz）表示，所以我的 SCollection 是 SCollection[Clazz] 类型而不是 SCollection[TableRow]。 TableRow 是 Java JDK 中的原生表示。关于如何降低内存分配的任何提示？它与我输入中的特定列类型有关：字节、字符串、记录、浮点数等？

【问题讨论】：

标签： google-bigquery google-cloud-dataflow apache-beam spotify-scio

【解决方案1】：

这可能是由于 TableRow 格式包含列的字符串名称，这增加了大小。

考虑using the following 创建对象的PCollection 而不是TableRows。这允许您直接读入与架构匹配的对象，这应该会稍微减少数据大小。

  /**
   * Reads from a BigQuery table or query and returns a {@link PCollection} with one element per
   * each row of the table or query result, parsed from the BigQuery AVRO format using the specified
   * function.
   *
   * <p>Each {@link SchemaAndRecord} contains a BigQuery {@link TableSchema} and a
   * {@link GenericRecord} representing the row, indexed by column name. Here is a
   * sample parse function that parses click events from a table.
   *
   * <pre>{@code
   * class ClickEvent { long userId; String url; ... }
   *
   * p.apply(BigQueryIO.read(new SerializableFunction<SchemaAndRecord, ClickEvent>() {
   *   public ClickEvent apply(SchemaAndRecord record) {
   *     GenericRecord r = record.getRecord();
   *     return new ClickEvent((Long) r.get("userId"), (String) r.get("url"));
   *   }
   * }).from("...");
   * }</pre>
   */
  public static <T> TypedRead<T> read(
      SerializableFunction<SchemaAndRecord, T> parseFn) {

【讨论】：

我正在使用一个类型安全的大查询案例类，它可能只是 scio 对类型读取而不是 TableRow 的包装
在内部，Dataflow 使用特定编码器并尝试对 PCollection 的采样元素进行编码以识别估计大小。这个编码器产生的输出是否可能比行的实际大小大得多？

【解决方案2】：

使用 Kryo 进行案例类序列化并不是最有效的，因为 Kryo 会序列化每个对象的完整类名。您可以通过注册经过扩展洗牌的类来解决此问题。

https://github.com/spotify/scio/wiki/FAQ#how-do-i-use-custom-kryo-serializers

这将在 Scio 0.7.x 中使用新的基于 Magnolia 宏的编码器派生解决。

https://github.com/spotify/scio/wiki/Coders

【讨论】：