如何比较Apache梁中两个PCollection中元素的相等性

Question

我试着编写了Apache K-意思是使用Java 编写的，但无法确定新的和旧的质量中心相等的终止条件，即PCollection是相等的。

1. 我尝试了Object.equals，但是结果显示为不相等，如何确定两个PCollection中的元素是相等的？

代码如下。

static void runWordCount(WordCountOptions options) {
        Pipeline p = Pipeline.create(options);
        PCollection<Double> p1=p.apply(Create.of(1.0,2.0,3.0));
        PCollection<Double> p2=p.apply(Create.of(1.0,2.0,3.0));
        boolean flag=p1.equals(p2);//return false
 }

1. 我目前的想法如下。

2.1。以数据点的形式读取文件：

PCollection<String> data

2.2。随机抽样K个数据点作为质量中心作为侧输入：

PCollectionView<String> center

2.3。向中心分配数据点: input PCollection<String> data output PCollection KV<String center,String data>

2.4.将PCollection KV<String center,String data>分组为KV<String center,Iterable<String data>>

2.5。使用附加输出(https://beam.apache.org/documentation/programming-guide/#output-tags)、输入KV<center,Iterable< data>>、输出<String center>和<Integer equal_flag>，计算新的质量中心并确定它是否与旧的质量中心相同。如果新旧素数相等，则为equal_flag=0，否则等于1

2.6。我想使用Sum将<Integer equal_flag> 中的数据合并到一个值中，然后确定它是否为0，从而结束循环。但是问题是，我无法获得PCollection中的元素(in PCollection，您可以使用.Collection获取RDD中的数据并判断它为)，，因为合并的数据仍然是PCollection，而对PCollectionView的转换要求我使用Dofn通过侧输入访问它。这样我就不能结束主程序的循环。

如果您有一个更好的想法来实现Apache means算法，请告诉我，非常感谢。

Answer 1

在梁模型中，没有直接的方法将PCollection中的值以一种独立于跑步者的方式获得到主程序。这是因为，对于任何分布式运行程序，您要查找的实际数据可能在远程工作人员中，而不是在提交作业的本地机器中。

要比较两个给定PCollection的值，一种方法可能是创建键和值相同的KVs，并执行加入操作。之后，后续的DoFn可以计数/筛选不兼容的值。

还有一个带有实用程序/断言的PAssert转换来比较PCollection，但是它主要用于测试，对您的场景可能没有用。