在 Apache Beam 中测试模式更新

Question

我正在创建一个无限制的 Dataflow 管道，并希望确保架构的新版本与旧版本兼容，以便可以在不停止的情况下对其进行更新。如果我的原始对象定义为：

@DefaultSchema(JavaFieldSchema.class)
public class TransactionPojoV1 {
  public final String bank;
  public final double purchaseAmount;

  @SchemaCreate
  public TransactionPojoV1(String bank, double purchaseAmount) {
    this.bank = bank;
    this.purchaseAmount = purchaseAmount;
  }
}

我想添加一个新字段fee：

@DefaultSchema(JavaFieldSchema.class)
public class TransactionPojoV2 {
  public final String bank;
  public final double purchaseAmount;
  public final double fee;

  @SchemaCreate
  public TransactionPojoV2(String bank, double purchaseAmount, double fee) {
    this.bank = bank;
    this.purchaseAmount = purchaseAmount;
    this.fee = fee
  }
}

我如何编写一个测试来测试TransactionPojoV2 是否可以从TransactionPojoV1 解码？并确保行为符合预期。

以上内容可能无法通过此测试，不确定，但我想要以下内容：

TransactionPojoV1 transactionPojoV1 = ...

byte[] encoded = Coder.encode(transactionPojoV1);

TransactionPojoV2 transactionPojoV2 = Coder.decode(encoded);

// Assert values are as expected.

我只是不知道该怎么做。

Answer 1

您可以从Pipeline 的SchemaRegistry 获取Coder。对于问题中给出的课程，您可以这样做：

@RunWith(JUnit4.class)
public class TransactionPojoTest {
    @Rule
    public TestPipeline p = TestPipeline.create();

    @Test
    public void testTransactionV1toV2() throws NoSuchSchemaException, IOException {
        SchemaRegistry schemaRegistry = p.getSchemaRegistry();

        Coder<TransactionPojoV1> transactionPojoV1Coder = schemaRegistry.getSchemaCoder(TransactionPojoV1.class);
        Coder<TransactionPojoV2> transactionPojoV2Coder = schemaRegistry.getSchemaCoder(TransactionPojoV2.class);

        TransactionPojoV1 inputTransaction = new TransactionPojoV1("Some Bank", 100.0);

        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        transactionPojoV1Coder.encode(inputTransaction, byteArrayOutputStream);

        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray());

        TransactionPojoV2 outputTransaction = transactionPojoV2Coder.decode(byteArrayInputStream);

        assertEquals(inputTransaction.bank, outputTransaction.bank);
        assertEquals(inputTransaction.purchaseAmount, outputTransaction.purchaseAmount, 1e-15);
        assertEquals(0D, outputTransaction.fee.doubleValue(), 1e-15);
    }
}

为此，TransactionPojoV2 需要：

@DefaultSchema(JavaFieldSchema.class)
public class TransactionPojoV2 {
    public final String bank;
    public final double purchaseAmount;
    @Nullable public final Double fee;

    @SchemaCreate
    public TransactionPojoV2(String bank, double purchaseAmount, @Nullable Double fee) {
        this.bank = bank;
        this.purchaseAmount = purchaseAmount;
        this.fee = fee != null ? fee : 0;
    }
}

Answer 2

Dataflow 运行器在这里可以更加宽松，因为它支持重新排序字段以确保更新兼容性。如果有新字段，它会将它们移动到 Schema 的“末尾”，并确保现有字段在新旧模式中的顺序相同。

这一切对您作为 Beam 用户来说都是透明的，这只是意味着只要您的新架构仅添加相对于旧架构的字段，它就应该是更新兼容的。

为了测试这种更新兼容性，我建议你使用这样的东西：

Schema v1 = schemaRegistry.getSchema(TransactionPojoV1.class);
Schema v2 = schemaRegistry.getSchema(TransactionPojoV2.class);

Set<Field> removedFields = Sets.difference(ImmutableSet.of(v1.getFields()), 
                                           ImmutableSet.of(v2.getFields()));
assertEmpty(removedFields);

请注意，这只是现成的，我还没有测试过。它还依赖于 guava 中的 Sets 和 ImmutableSet。

另外，值得注意的是，Dataflow 在实践中可能比这更宽松。 Dataflow 只关心融合边界（即存在 GroupByKey/shuffle 的地方）的更新兼容性。