如何解决使用 Beam Dataflow 和 Put 进行单元测试总是失败
我正在使用 Apache Beam Dataflow 创建一个管道,该管道从 GCP PubSub 读取并写入 BigTable 实例中的表。
现在我正在尝试构建一个单元测试,以检查接收到的数据是否已很好地转换为我的 BigTable 架构。问题是我正在构建的单元测试(我觉得它应该非常简单)在将生成的突变与预期的突变进行比较时总是失败。
Caused by: java.lang.AssertionError:
Expected: iterable over [<KV{name:value,[{"totalColumns":1,"row":"name:value","families":{"F":[{"qualifier":"name","vlen":5,"tag":[],"timestamp":"1"}]}}]}>] in any order
but: Not matched: <KV{name:value,"timestamp":"1"}]}}]}>
我做了一个非常简单的测试来证明这一点,但由于上述错误而失败。
package com.energyworx.dataflow.transforms.integration;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import java.util.Collections;
public class SimpleTest {
@Rule
public final transient TestPipeline pipeline = TestPipeline.create();
private static final byte[] COLUMN_FAMILY_BYTES = Bytes.toBytes("F");
private static final String INPUT_VALUE = "name:value";
private static class TestDoFn extends DoFn<String,KV<String,Iterable<Mutation>>> {
@ProcessElement
public void processElement(ProcessContext c) {
String element = c.element();
Mutation row = createRow(element);
c.output(KV.of(element,Collections.singletonList(row)));
}
}
@Test
@Category(NeedsRunner.class)
public void testMutations() {
PCollection<KV<String,Iterable<Mutation>>> output = pipeline
.apply(Create.of(INPUT_VALUE))
.apply(ParDo.of(new TestDoFn()));
Mutation row = createRow(INPUT_VALUE);
PAssert.that(output).containsInAnyOrder(KV.of(INPUT_VALUE,Collections.singletonList(row)));
pipeline.run();
}
static Mutation createRow(String rowKey) {
Put row = new Put(Bytes.toBytes(rowKey));
String[] parts = rowKey.split(":");
row.addColumn(COLUMN_FAMILY_BYTES,Bytes.toBytes(parts[0]),1L,Bytes.toBytes(parts[1]));
return row;
}
}
由于这失败了,我尝试了一些不同的方法,例如创建一个新的 DoFn 来接收预期值并针对管道值进行断言,并尝试向我们 PAssert.that(output).satisfies(expectedValues)
并编写自定义函数,但这两种方法失败 Caused by: java.io.NotSerializableException: org.apache.hadoop.hbase.client.Put
。
我觉得这是一个非常简单的用例,我遗漏了一些东西或者我做错了什么。
关于我做错了什么的任何提示?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。