util/kvstore/KVStoreSerializer.java

0001 /*
0002  * Licensed to the Apache Software Foundation (ASF) under one or more
0003  * contributor license agreements.  See the NOTICE file distributed with
0004  * this work for additional information regarding copyright ownership.
0005  * The ASF licenses this file to You under the Apache License, Version 2.0
0006  * (the "License"); you may not use this file except in compliance with
0007  * the License.  You may obtain a copy of the License at
0008  *
0009  *    http://www.apache.org/licenses/LICENSE-2.0
0010  *
0011  * Unless required by applicable law or agreed to in writing, software
0012  * distributed under the License is distributed on an "AS IS" BASIS,
0013  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014  * See the License for the specific language governing permissions and
0015  * limitations under the License.
0016  */
0017
0018 package org.apache.spark.util.kvstore;
0019
0020 import java.io.ByteArrayInputStream;
0021 import java.io.ByteArrayOutputStream;
0022 import java.util.zip.GZIPInputStream;
0023 import java.util.zip.GZIPOutputStream;
0024 import static java.nio.charset.StandardCharsets.UTF_8;
0025
0026 import com.fasterxml.jackson.databind.ObjectMapper;
0027
0028 import org.apache.spark.annotation.Private;
0029
0030 /**
0031  * Serializer used to translate between app-defined types and the LevelDB store.
0032  *
0033  * <p>
0034  * The serializer is based on Jackson, so values are written as JSON. It also allows "naked strings"
0035  * and integers to be written as values directly, which will be written as UTF-8 strings.
0036  * </p>
0037  */
0038 @Private
0039 public class KVStoreSerializer {
0040
0041   /**
0042    * Object mapper used to process app-specific types. If an application requires a specific
0043    * configuration of the mapper, it can subclass this serializer and add custom configuration
0044    * to this object.
0045    */
0046   protected final ObjectMapper mapper;
0047
0048   public KVStoreSerializer() {
0049     this.mapper = new ObjectMapper();
0050   }
0051
0052   public final byte[] serialize(Object o) throws Exception {
0053     if (o instanceof String) {
0054       return ((String) o).getBytes(UTF_8);
0055     } else {
0056       ByteArrayOutputStream bytes = new ByteArrayOutputStream();
0057       try (GZIPOutputStream out = new GZIPOutputStream(bytes)) {
0058         mapper.writeValue(out, o);
0059       }
0060       return bytes.toByteArray();
0061     }
0062   }
0063
0064   @SuppressWarnings("unchecked")
0065   public final <T> T deserialize(byte[] data, Class<T> klass) throws Exception {
0066     if (klass.equals(String.class)) {
0067       return (T) new String(data, UTF_8);
0068     } else {
0069       try (GZIPInputStream in = new GZIPInputStream(new ByteArrayInputStream(data))) {
0070         return mapper.readValue(in, klass);
0071       }
0072     }
0073   }
0074
0075   final byte[] serialize(long value) {
0076     return String.valueOf(value).getBytes(UTF_8);
0077   }
0078
0079   final long deserializeLong(byte[] data) {
0080     return Long.parseLong(new String(data, UTF_8));
0081   }
0082
0083 }