java - Apache 光束中的 Gamma 分布

Question

我正在尝试在 apache beam 中实现 Gamma Distribution。首先，我正在使用 Apache Beam 的 TextIO 类读取 CSV 文件 CSV 文件:

Pipeline p = Pipeline.create();  
p.apply(TextIO.read().from("gs://path/to/file.csv"));  

之后，我应用一个转换来解析 CSV 文件中的每一行并返回一个对象。我只是在这里尝试执行 Gamma 分布操作:

.apply(ParDo.of(new DoFn<String, Entity>() {
@ProcessElement
public void processElement(ProcessContext c) {
    String[] strArr = c.element().split(",");
    ClassxNorms xn = new ClassxNorms();
    xn.setDuration(Double.parseDouble(strArr[0]));
    xn.setAlpha(Double.parseDouble(strArr[1]));
    xn.setBeta(Double.parseDouble(strArr[2]));
    GammaDistribution gdValue = new GammaDistribution(Double.parseDouble(strArr[0]), Double.parseDouble(strArr[1]), Double.parseDouble(strArr[2]));
    System.out.println("gdValue : " + gdValue);
    c.output(xn);
}
}));

我正在创建一个 BeamRecord，下一步我会将 Beam 记录转换为字符串，以将最终输出写入 Google 存储:

PCollection<String> gs_output_final = xnorm_trig.apply(ParDo.of(new DoFn<BeamRecord, String>() {
                    private static final long serialVersionUID = 1L;
                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        c.output(c.element().toString());
                        System.out.println(c.element().toString());
                    }
                })); 
   gs_output_final.apply(TextIO.write().to("gs://output/op_1/Q40test111"));  

我得到了输出，但 Gamma 分布操作没有得到实现。任何帮助将非常感激。

Answer 1

我能够在 apache beam 中实现 Gamma 分布。以下是供引用的代码片段:

.apply(ParDo.of(new DoFn<String, ClassxNorms>() { 
    @ProcessElement
    public void processElement(ProcessContext c) throws ParseException {
      String[] strArr = c.element().split(",");
      ClassxNorms xn = new ClassxNorms();
      double sample = new GammaDistribution(Double.parseDouble(strArr[11]), Double.parseDouble(strArr[12])).cumulativeProbability(Double.parseDouble(strArr[6]));
      xn.setDuration(Double.parseDouble(strArr[6]));
      xn.setAlpha(Double.parseDouble(strArr[11]));
      xn.setBeta(Double.parseDouble(strArr[12]));
      xn.setVolume(Double.parseDouble(strArr[13]));
      xn.setSpend(Double.parseDouble(strArr[14]));
      xn.setEfficiency(Double.parseDouble(strArr[15]));
      xn.setXnorm(Double.parseDouble(strArr[16]));
      xn.setKey(strArr[17]);
      xn.setGamma(sample);
      c.output(xn);
    }
  }));