flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Kirschnick, Johannes" <johannes.kirschn...@tu-berlin.de>
Subject AW: Operating on Serialized Data
Date Wed, 25 Feb 2015 09:04:46 GMT
Hi Max,

thanks for the detailed answer-
That was exactly what I have been looking for.
I switched the serialization from kryo to use the Value interface instead, keeping everything
constant basically halved the execution time - nice.

One note - iterating over the array in serial fashion works, but smells inefficient.
Would it make sense to do bulk reading using some byte buffer - or is that not needed as the
DataInputView is backed by the memory manager and that would cause issues?

Johannes
________________________________________
Von: Max Michels <mxm@apache.org>
Gesendet: Dienstag, 24. Februar 2015 18:11
An: dev@flink.apache.org
Betreff: Re: Operating on Serialized Data

Apparently, the mailing list doesn't allow attachments.

Here the example with syntax highlighting:
https://gist.github.com/mxm/d1929b4b69dda87d5c37


public class CustomSerializer {

   public static class Vector implements Value {

      private transient double[] doubleValues;

      public Vector() {
      }

      public Vector(double[] doubleValues) {
         this.doubleValues = doubleValues;
      }

      public double getElement(int position) {
         return doubleValues[position];
      }

      public void setElement(double value, int position) {
         doubleValues[position] = value;
      }

      public void multiply(int factor) {
         for (int i = 0; i < doubleValues.length; i++) {
            doubleValues[i] *= factor;
         }
      }

      @Override
      public void write(DataOutputView out) throws IOException {
         out.writeInt(doubleValues.length);
         for (double value : doubleValues) {
            out.writeDouble(value);
         }
      }

      @Override
      public void read(DataInputView in) throws IOException {
         int length = in.readInt();
         double[] array = new double[length];
         for (int i = 0; i < length; i++) {
            array[i] = in.readDouble();
         }
         this.doubleValues = array;
      }

      @Override
      public String toString() {
         return "Vector{" +
               "doubleValues=" + Arrays.toString(doubleValues) +
               '}';
      }
   }

   public static void main(String[] args) throws Exception {
      ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

      Vector[] vectorList = new Vector[1024];

      // create some sample data
      for (int v = 0; v < vectorList.length; v++) {
         double[] arr = new double[128];
         for (int i = 0; i < arr.length; i++) {
            arr[i] = i * 1.23 * v;
         }
         vectorList[v] = new Vector(arr);
      }

      // create data set
      DataSet<Vector> source = env.fromElements(vectorList);

      // multiply all vectors by 2
      DataSet<Vector> ds = source.map(new MapFunction<Vector, Vector>() {
         private static final long serialVersionUID = -1511665386949403921L;

         @Override
         public Vector map(Vector value) throws Exception {
            value.multiply(2);
            return value;
         }
      });

      ds.print();

      env.execute();

   }
}


On Tue, Feb 24, 2015 at 5:43 PM, Max Michels <mxm@apache.org> wrote:
> Hi Johannes,
>
> Thanks for your question. You can try to implement the Value interface
> for your Vector PoJo. It has to have an empty constructor and
> implement the write and read methods of the interface for
> serialization.
>
> Based on your description, I've implemented an example to demonstrate
> the use of the Value interface. It would be interesting to hear from
> you whether you could decrease the serialization time using this
> serialization method.
>
> Best regards,
> Max
>
>
> On Tue, Feb 24, 2015 at 11:13 AM, Kirschnick, Johannes
> <johannes.kirschnick@tu-berlin.de> wrote:
>> Hi list,
>>
>>
>> I have a general question on as to whether it's possible to significantly speed up
the processing by cutting down on the serialization costs during iterations.
>>
>>
>> The basic setup that I have are a couple of vectors that are repeatedly mutated (added
& multiplied) as part of an iterative run within a reducer.
>>
>> A vector is basically "just" an array of doubles - all of the same size.
>>
>>
>> I noticed during simple profiling that roughly 50% of the execution time is spent
on serializing the data in using the com.esotericsoftware.kryo.serializers.DefaultArraySerializers
in Kryo.
>>
>>
>> I know that any custom operation would would varant custom processing, but given
the serialization contributes such a large amount of processing time to the overall runtime
it might very well be worthwhile
>>
>>
>> Is that currently exposed in any fashion to the user code, or are there some hooks
I could look into?
>>
>>
>> Thanks
>>
>> Johannes
Mime
View raw message