知識庫 / JSON RSS 訂閱

1. 簡介

本教程將介紹如何在 Java 中將 JSON 數據轉換為 Apache Avro 對象。 Avro 是一個數據序列化框架，它提供豐富的結構化數據和緊湊的二進制數據。 此外，與其它序列化框架不同，Avro 使用在 JSON 格式中定義的模式，而不是要求對序列化進行代碼生成。

作為結果，它的一個關鍵優勢是模式演進的支持。 這樣，Avro 特別適合於需要處理隨時間變化的數據結構的應用程序。 此外，由於其緊湊的數據格式，它也適用於處理大量數據的應用程序。

2. JSON 到 Avro 轉換

在 Avro 中，從 JSON 轉換為對象需要一個 schema 來定義數據結構以及一個轉換機制。 在我們的情況下，這個轉換機制將位於 convertJsonToAvro() 方法中。

schema 定義了數據的格式（包括字段名稱和類型），而該方法利用該 schema 將 JSON 轉換為 Avro 對象。

2.1. 實現轉換方法

首先，我們需要將必要的依賴項添加到我們的 <em pom.xml:</em>：

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.12.0</version>
</dependency>

接下來，讓我們創建一個模式，定義我們的JSON應該遵循的結構：

private static final String SCHEMA_JSON = """
    {
        "type": "record",
        "name": "Customer",
        "namespace": "com.baeldung.avro",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": "int"},
            {"name": "email", "type": ["null", "string"], "default": null}
        ]
    }""";

接下來，我們創建一個處理 JSON 到 Avro 轉換的方法。此外，轉換過程涉及三個主要組件：模式、一個根據模式讀取 JSON 數據的解碼器，以及一個用於創建 Avro 對象的DatumReader。

讓我們創建一個方法：

GenericRecord convertJsonToAvro(String json) throws IOException {
    
    try {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        Decoder decoder = DecoderFactory.get().jsonDecoder(schema, json);
        return reader.read(null, decoder);
    } catch (IOException e) {
        throw new IOException("Error converting JSON to Avro", e);
    }
}

使用我們實例化過的解碼器，我們讀取 JSON 輸入並確保它與我們的模式結構相匹配。 最終，我們使用 DatumReader，利用模式和解碼器創建 GenericRecord 對象。 這樣 Avro 能夠以不生成類的方式表示數據。

在將 JSON 轉換為 Avro 時，我們通常遵循以下步驟：

1. JSON 輸入與模式進行驗證
2. 解碼器根據模式的結構解析 JSON
3. DatumReader 創建一個包含數據的 GenericRecord 對象
4. 如果 JSON 中不存在但模式中定義的字段，則分配它們的默認值

需要注意的是，Avro 如何處理聯合類型。 當我們定義一個字段為聯合類型（例如：["null", "string"]），JSON 表示必須明確指定正在使用的類型。 例如，我們必須將字符串值包裝在一個鍵為類型的 JSON 對象中：{"string": "value"}。 這樣與常規 JSON 不同，我們只是直接使用值。

2.2. 測試 JSON 到 Avro 轉換

現在，讓我們測試我們的實現：

@Test
void whenValidJsonInput_thenConvertsToAvro() throws IOException {
    
    JsonToAvroConverter converter = new JsonToAvroConverter();
    String json = "{\"name\":\"John Doe\",\"age\":30,\"email\":{\"string\":\"[email protected]\"}}";

    GenericRecord record = converter.convertJsonToAvro(json);

    assertEquals("John Doe", record.get("name").toString());
    assertEquals(30, record.get("age"));
    assertEquals("[email protected]", record.get("email").toString());
}

本測試驗證我們的轉換器能夠正確處理完整的 JSON 對象（所有字段均已填充）。此外，請注意 email  字段的特殊格式。該格式使用了 Avro 的聯合類型語法。

如測試所示，所有類型均已正確轉換並可在 GenericRecord 變量 result  中訪問。

讓我們來查看我們的下一個測試，其中我們將一個包含 null 字段的 JSON 轉換為 GenericRecord：

@Test
void whenJsonWithNullableField_thenConvertsToAvro() throws IOException {
    
    JsonToAvroConverter converter = new JsonToAvroConverter();
    String json = "{\"name\":\"John Doe\",\"age\":30,\"email\":null}";

    GenericRecord record = converter.convertJsonToAvro(json);

    assertEquals("John Doe", record.get("name").toString());
    assertEquals(30, record.get("age"));
    assertNull(record.get("email"));
}

本測試確認我們已正確地轉換了可選的 電子郵件 字段中的 空值 。我們已將 電子郵件 字段定義為 [null , 字符串 ] 的並集，因此它接受 null 值。

3. 高級用法

通過將 JSON 轉換為 Avro 對象，我們已經涵蓋了許多常見情況。然而，實際應用通常需要更復雜的操作。

下面我們來看兩個場景：處理 JSON 數組（對於數據集處理至關重要）和二進制序列化。後者為數據存儲或通信做準備。

3.1. 處理 JSON 數組

有時，我們需要同時處理多個 JSON 對象。 考慮到這一點，讓我們擴展我們的轉換器以處理 JSON 數組。為此，讓我們創建一個新的方法：

List<GenericRecord> convertJsonArrayToAvro(String jsonArray) throws IOException {
    
    List<GenericRecord> records = new ArrayList<>();
    
    Schema arraySchema = Schema.createArray(schema);
    
    Decoder decoder = DecoderFactory.get().jsonDecoder(arraySchema, jsonArray);
    DatumReader<List<GenericRecord>> reader = new GenericDatumReader<>(arraySchema);
    
    List<GenericRecord> result = reader.read(null, decoder);
    return result;
}

現在，讓我們分析我們的方法。首先，我們正在創建一個現有記錄模式的數組模式。接下來，我們使用 Avro 的內置 JSON 解碼器來驗證 JSON (“arraySchema”) 是否符合模式定義的結構，將每個字段轉換為其模式等效項，然後相應地處理諸如聯合類型之類的特殊情況。

@Test
void whenJsonArray_thenConvertsToAvroList() throws IOException {
    
    JsonToAvroConverter converter = new JsonToAvroConverter();
    String jsonArray = """
        [
            {"name":"John Doe","age":30,"email":{"string":"[email protected]"}},
            {"name":"Jane Doe","age":28,"email":{"string":"[email protected]"}}
        ]""";

    List<GenericRecord> records = converter.convertJsonArrayToAvro(jsonArray);

    assertEquals(2, records.size());
    assertEquals("John Doe", records.get(0).get("name").toString());
    assertEquals("[email protected]", records.get(1).get("email").toString());
}

byte[] serializeAvroRecord(GenericRecord record) throws IOException {
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);
    
    writer.write(record, encoder);
    encoder.flush();
    return outputStream.toByteArray();
}

@Test
void whenSerializingAvroRecord_thenProducesByteArray() throws IOException {
    String json = """ 
        {"name":"John Doe","age":30,"email":{"string":"[email protected]"}}
        """ ;
    JsonToAvroConverter converter = new JsonToAvroConverter();
    GenericRecord record = converter.convertJsonToAvro(json);

    byte[] bytes = converter.serializeAvroRecord(record);

    assertNotNull(bytes);
    assertTrue(bytes.length > 0);
}