我正在研究通过Google Dataflow / Apache Beam处理来自Web用户会话的日志，并且需要将用户日志的输入（流式传输）与上个月用户会话的历史记录结合起来。
我研究了以下方法：
element的processElement(ProcessContext processContext)我的理解是，通过加载的数据.withSideInputs(pCollectionView)需要放入内存中。我知道我可以将单个用户的所有会话历史记录都存储到内存中，但不能将所有会话历史记录都存储到内存中。
我的问题是，是否有一种方法可以从仅与当前用户会话相关的侧面输入中加载/流式传输数据？
我正在想象一个parDo函数，该函数将通过指定用户的ID从侧面输入中加载用户的历史会话。但是只有当前用户的历史记录会话可以容纳在内存中。通过侧面输入加载所有历史记录会话将太大。
一些伪代码来说明：
public static class MetricFn extends DoFn<LogLine, String> {
final PCollectionView<Map<String, Iterable<LogLine>>> pHistoryView;
public MetricFn(PCollectionView<Map<String, Iterable<LogLine>>> historyView) {
this.pHistoryView = historyView;
}
@Override
public void processElement(ProcessContext processContext) throws Exception {
Map<String, Iterable<LogLine>> historyLogData = processContext.sideInput(pHistoryView);
final LogLine currentLogLine = processContext.element();
final Iterable<LogLine> userHistory = historyLogData.get(currentLogLine.getUserId());
final String outputMetric = calculateMetricWithUserHistory(currentLogLine, userHistory);
processContext.output(outputMetric);
}
}
目前尚没有一种方法可以访问流中的每个键的侧面输入，但是正如您所描述的，它绝对有用，这是我们正在考虑实现的方法。
一种可能的解决方法是使用侧面输入来分配指向实际会话历史记录的指针。生成24小时会话历史记录的代码可以将它们上传到GCS / BigQuery / etc，然后将位置作为侧面输入发送到加入代码。
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句