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

Question

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

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

Answer 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) {

Answer 2

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

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