无法通过 FileIO.Write/WriteFiles/WriteShardedBundlesToTempFiles/GroupIntoShards

Question

我实现了从 Pub/Sub 读取包含 AWS S3 密钥的 JSON 的管道；并下载由换行符分隔的 JSON (NDJson) 组成的 S3 对象；然后，使用窗口和 GroupByKey 将 NDJson 写入 GCS 中的文件。

显然，它正在以 10 秒的窗口和AfterPane.elementCountAtLeast(1000). 进行流式传输，但是，窗口没有被分割，并且 GCS 中没有文件。此外，还发生了异常Shutting down JVM after 8 consecutive periods of measured GC thrashing. Memory is used/total/max = 4295/4946/4946 MB, GC last/max = 98.54/98.54 %, #pushbacks=1, gc thrashing=true. Heap dump not written.。

下图是我的 Dataflow 的作业图。 GroupIntoShards 不输出数据。我不明白为什么会这样。此外，这是发生未绑定集。换句话说，如果 Pub/Sub 消息的数量很少（可能不发生 GC 就足够了），结果是好的。原因可能是关闭窗口，因为 Pub/Sub 的订阅没有任何未确认的消息。但是，如果消息被连续放置，GroupIntoShards 不起作用并且 GCS 中没有文件。 enter image description here

以下是我的部分来源：

在主类中，

public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(options);

    PCollection<String> inputed =
        .apply("Read PubSub Events", PubsubIO.readMessagesWithAttributes().fromTopic(options.getInputTopic()))
        .apply("NDJson Divider", ParDo.of(new NDJsonDivider(options.getSecretId())));

    inputed
        .apply(options.getWindowDuration() + " Window",
            Window.<String>into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration())))
                .triggering(
                    Repeatedly.forever(
                        AfterFirst.of(
                            AfterWatermark.pastEndOfWindow(),
                            AfterPane.elementCountAtLeast(1000),
                            AfterProcessingTime.pastFirstElementInPane().plusDelayOf(DurationUtils.parseDuration(options.getWindowDuration())))))
                .accumulatingFiredPanes()
                .withAllowedLateness(Duration.standardDays(2))
        )
        .apply(FileIO.<String, String>writeDynamic()
            .by(new TableRowPartitionContextFn())
            .via(TextIO.sink())
            .to(options.getOutputDirectory())
            .withNaming(PartitionedFileNaming::new)
            .withNumShards(options.getNumShards())
            .withDestinationCoder(StringUtf8Coder.of())
            .withCompression(Compression.GZIP)
        );

    return pipeline.run();
  }

在NDJsonDivider 类中，

  @ProcessElement
  public void processElement(ProcessContext c) {
    JsonObject msg;
    try {
      msg = new JsonParser().parse(new String(c.element().getPayload())).getAsJsonObject();
    } catch (JsonSyntaxException | IllegalStateException e) {
      LOG.error(e);
      return;
    }

    if (msg.has(RECORDS) && msg.get(RECORDS).isJsonArray()) {
      JsonArray records = msg.get(RECORDS).getAsJsonArray();
      try (BufferedReader reader = processPutEventMessage(records)) {
        if (reader == null) {
          return;
        }
        String line;
        while (Objects.nonNull(line = reader.readLine())) {
          c.outputWithTimestamp(line, Instant.now());
        }
      } catch (IOException e) {
        LOG.error("Failed to process Put Event Message: ", e);
      }
    }
  }

  private BufferedReader processPutEventMessage(JsonArray records) {
    try {
      putEventMessage s3Obj = extractS3ObjectInfo(records);
      GCPBridge.getInstance().setSecretId(this.secretId);
      return S3.getReader(s3Obj.region, s3Obj.bucket, s3Obj.key, "UTF-8");
    } catch (IllegalArgumentException | AWSS3Exception | NoSuchKeyException e) {
      LOG.error("Failed to access S3:", e);
    }
    return null;
  }

在TableRowPartitionContextFn 类中，

class TableRowPartitionContextFn implements SerializableFunction<String, String> {
  @Override
  public String apply(String e) {
    JsonObject data = new JsonParser().parse(e).getAsJsonObject();
    String Id = "", Type = "";
    if (data.has("id")) {
      Id = data.get("id").getAsString();
    }
    if (data.has("type")) {
      Type = data.get("type").getAsString();
    }
    return Id + "/" + Type;
  }
}

mvn -Pdataflow-runner compile exec:java -Dexec.mainClass=com.dev.playground -Dexec.args="--project=PROJECTID --inputTopic=projects/PROJECTID/topics/dev --outputDirectory=gs://dev/playground/output/ --secretId=dev --tempLocation=gs://dev/playground/tmp/ --runner=DataflowRunner

默认设置持续时间 @Default.String("10s") String getWindowDuration();

Answer 1

我怀疑Repeatedly.forever() 与.accumulatingFiredPanes() 的组合将永远存储数据的缓冲区，最终导致内存不足错误。

请尝试使用discardingFiredPanes()，它不会保留先前触发窗格的元素。

这意味着只有到达的新元素才会在触发时在窗格中发出。您似乎没有对窗格中的所有值进行任何类型的聚合/组合结果。看起来您只是将它们直接写入输出，所以我认为您不需要 accumulatingFiredPanes。

如果您想根据窗口中的所有元素（例如平均值、总和等）计算某种聚合统计数据或值。然后我会建议在窗口之后使用组合器来累积结果。