带有beam数据流和put的java单元测试总是失败

frebpwbc  于 2021-07-13  发布在  Hbase
关注(0)|答案(0)|浏览(360)

我正在使用apachebeam数据流创建一个管道,该管道从gcp pubsub读取数据,然后写入bigtable示例中的表。
现在,我正在尝试构建一个单元测试,检查接收到的数据是否已很好地转换为bigtable模式。问题是,我正在构建的单元测试(我觉得应该是非常直接的)在将生成的突变与预期的突变进行比较时总是失败。

  1. Caused by: java.lang.AssertionError:
  2. Expected: iterable over [<KV{name:value, [{"totalColumns":1,"row":"name:value","families":{"F":[{"qualifier":"name","vlen":5,"tag":[],"timestamp":"1"}]}}]}>] in any order
  3. but: Not matched: <KV{name:value, [{"totalColumns":1,"row":"name:value","families":{"F":[{"qualifier":"name","vlen":5,"tag":[],"timestamp":"1"}]}}]}>

我做了一个非常简单的测试来证明这一点,它失败了上面的错误。

  1. package com.energyworx.dataflow.transforms.integration;
  2. import org.apache.beam.sdk.testing.NeedsRunner;
  3. import org.apache.beam.sdk.testing.PAssert;
  4. import org.apache.beam.sdk.testing.TestPipeline;
  5. import org.apache.beam.sdk.transforms.Create;
  6. import org.apache.beam.sdk.transforms.DoFn;
  7. import org.apache.beam.sdk.transforms.ParDo;
  8. import org.apache.beam.sdk.values.KV;
  9. import org.apache.beam.sdk.values.PCollection;
  10. import org.apache.hadoop.hbase.client.Mutation;
  11. import org.apache.hadoop.hbase.client.Put;
  12. import org.apache.hadoop.hbase.util.Bytes;
  13. import org.junit.Rule;
  14. import org.junit.Test;
  15. import org.junit.experimental.categories.Category;
  16. import java.util.Collections;
  17. public class SimpleTest {
  18. @Rule
  19. public final transient TestPipeline pipeline = TestPipeline.create();
  20. private static final byte[] COLUMN_FAMILY_BYTES = Bytes.toBytes("F");
  21. private static final String INPUT_VALUE = "name:value";
  22. private static class TestDoFn extends DoFn<String, KV<String, Iterable<Mutation>>> {
  23. @ProcessElement
  24. public void processElement(ProcessContext c) {
  25. String element = c.element();
  26. Mutation row = createRow(element);
  27. c.output(KV.of(element, Collections.singletonList(row)));
  28. }
  29. }
  30. @Test
  31. @Category(NeedsRunner.class)
  32. public void testMutations() {
  33. PCollection<KV<String, Iterable<Mutation>>> output = pipeline
  34. .apply(Create.of(INPUT_VALUE))
  35. .apply(ParDo.of(new TestDoFn()));
  36. Mutation row = createRow(INPUT_VALUE);
  37. PAssert.that(output).containsInAnyOrder(KV.of(INPUT_VALUE, Collections.singletonList(row)));
  38. pipeline.run();
  39. }
  40. static Mutation createRow(String rowKey) {
  41. Put row = new Put(Bytes.toBytes(rowKey));
  42. String[] parts = rowKey.split(":");
  43. row.addColumn(COLUMN_FAMILY_BYTES, Bytes.toBytes(parts[0]), 1L, Bytes.toBytes(parts[1]));
  44. return row;
  45. }
  46. }

由于这是失败的,我尝试了一些不同的方法,比如创建一个新的dofn,它接收预期的值和针对管道值的Assert,并尝试使用 PAssert.that(output).satisfies(expectedValues) 以及编写自定义函数,但这两个函数都失败了 Caused by: java.io.NotSerializableException: org.apache.hadoop.hbase.client.Put .
我觉得这是一个非常简单的用例,我遗漏了一些东西或者我做错了什么。
关于我做错了什么有什么建议吗?

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题