如何在Avro中定义LogicalType. (java) [英] How to define a LogicalType in Avro. (java)

问题描述

我需要能够在AVRO模式中标记一些字段，以便在序列化时对它们进行加密.

I need to be able to mark some fields in the AVRO schema so that they will be encrypted at serialization time.

logicalType允许标记字段，并且与自定义转换一起应允许AVRO透明地对其加密.

A logicalType allows to mark the fields, and together with a custom conversion should allow to let them be encrypted transparently by AVRO.

我在查找有关如何在AVRO中定义和使用新的logicalType的文档时遇到了一些问题(

I had some issues to find documentation on how to define and use a new logicalType in AVRO (avro_1.8.2#Logical+Types).
I decided then to share here in the answer what I found, to easy the life of anyone else getting on it and to get some feedback in case I'm doing something wrong.

推荐答案

首先，我将逻辑类型定义为:

First of all I defined a logicalType as:

public class EncryptedLogicalType extends LogicalType {
    //The key to use as a reference to the type
    public static final String ENCRYPTED_LOGICAL_TYPE_NAME = "encrypted";

    EncryptedLogicalType() {
        super(ENCRYPTED_LOGICAL_TYPE_NAME);
    }

    @Override
    public void validate(Schema schema) {
        super.validate(schema);
        if (schema.getType() != Schema.Type.BYTES) {
            throw new IllegalArgumentException(
                    "Logical type 'encrypted' must be backed by bytes");
        }
    }
}

然后进行新的转换:

public class EncryptedConversion extends Conversion<ByteBuffer> {
    // Construct a unique instance for all the conversion. This have to be changed in case the conversion
    //   needs some runtime information (e.g.: an encryption key / a tenant_ID). If so, the get() method should 
    //   return the appropriate conversion per key.
    private static final EncryptedConversion INSTANCE = new EncryptedConversion();
    public static final EncryptedConversion get(){ return INSTANCE; }
    private EncryptedConversion(){ super(); }

    //This conversion operates on ByteBuffer and returns ByteBuffer
    @Override
    public Class<ByteBuffer> getConvertedType() { return ByteBuffer.class; }

    @Override
    public String getLogicalTypeName() { return EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME; }

    // fromBytes and toBytes have to be overridden as this conversion works on bytes. Other may need to be 
    //  overridden. The types supported need to be updated also in EncryptedLogicalType#validate(Schema schema)
    @Override
    public ByteBuffer fromBytes(ByteBuffer value, Schema schema, LogicalType type) {
        encryptedValue = __encryptionLogic__(value); 
        return encryptedValue;
    }

    @Override
    public ByteBuffer toBytes(ByteBuffer value, Schema schema, LogicalType type) {
        decryptedValue = __decryptionLogic__(value); 
        return decryptedValue;
    }
}

.avsc模式文件将类似于:

The .avsc schema file will be similar to:

{
    "name": "MyMessageWithEncryptedField",
    "type": "record",
    "fields": [
        {"name": "payload","type" : {"type" : "bytes","logicalType" : "encrypted"}},
        ...

最后，在从架构文件生成的MyMessageWithEncryptedField.java类中，我添加了返回转换的方法:

Finally in the MyMessageWithEncryptedField.java class generated out of the schema file I added the method to return the conversion:

@Override
public Conversion<?> getConversion(int fieldIndex) {
    // This allow us to have a more flexible conversion retrieval, so we don't have to code it per field.
    Schema fieldSchema = SCHEMA$.getFields().get(fieldIndex).schema();
    if ((fieldSchema.getLogicalType() != null)
            && (fieldSchema.getLogicalType().getName() == EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME)){
     // here we could pass to the get() method a runtime information, e.g.: a tenantId that can be found in the data structure.
        return EncryptedConversion.get();
    }
    return null;
}

要使其运行，我仍然必须在运行时注册类型:

To make it run I still have to register the type at runtime:

LogicalTypes.register(EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME, new LogicalTypes.LogicalTypeFactory() {
    private final LogicalType encryptedLogicalType = new EncryptedLogicalType();
    @Override
    public LogicalType fromSchema(Schema schema) {
        return encryptedLogicalType;
    }
});

一些注意事项:

• 如果您的logicalType需要从架构定义中传入其他一些属性，则可以从avro.lang.java.avro.src.main.java.org.apache.avro.LogicalTypes.Decimal
• 最后一段代码(寄存器)当前在我的逻辑开始之前运行，但我计划将其移动到架构生成的类(MyMessageWithEncryptedField.java)内的静态块中

• if your logicalType needs some other properties passed in from the schema definition, you can modify the LogicalType class taking example from avro.lang.java.avro.src.main.java.org.apache.avro.LogicalTypes.Decimal
• the last piece of code (the register) is currently run before my logic starts, but I plan to move it in a static block inside the schema generated class (MyMessageWithEncryptedField.java)

这篇关于如何在Avro中定义LogicalType. (java)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！