Pfthb

I am trying to get familiar with the semantics of Flink after having started with Spark. I would like to write a DataSet[IndexNode] to persistent storage in HDFS so that it can be read later by another process. Spark has a simple ObjectFile API that provides such a functionality, but I cannot find a similar option in Flink.

case class IndexNode(vec: Vector[IndexNode],
 id: Int) extends Serializable 
 // Getters and setters etc. here

The build-in sinks tend to serialize the instance based on the toString method, which is not suitable here due to the nested structure of the class. I imagine the solution is to use a FileOutputFormat and provide a translation of the instances to a byte stream. However, I am not sure how to serialize the vector, which is of an arbitrary length and can be many levels deep.

You can achieve this by using SerializedOutputFormat and SerializedInputFormat.

Try following steps:

1. 
   

   Make IndexNode extend IOReadableWritable interface from FLINK. Make unserialisable fields @transient. Implement write(DataOutputView out) and read(DataInputView in) method. The write method will write out all data from IndexNode and read method will read them back and build all internal data fields. For example, instead of serialising all data from arr field in Result class, I write out all value out and then read them back and rebuild the array in read method.

   
   
   

   class Result(var name: String, var count: Int) extends IOReadableWritable 
   
    @transient
    var arr = Array(count, count)
   
    def this() 
    this("", 1)
    
   
    override def write(out: DataOutputView): Unit = 
    out.writeInt(count)
    out.writeUTF(name)
    
   
    override def read(in: DataInputView): Unit = 
    count = in.readInt()
   
    name = in.readUTF()
   
    arr = Array(count, count)
    
   
    override def toString: String = s"$name, $count, $getArr"
   
   
   

   
   


2. 
   

   Write out data with

   
   
   

   myDataSet.write(new SerializedOutputFormat[Result], "/tmp/test")
   

   
   
   

   and read it back with

   
   
   

   env.readFile(new SerializedInputFormat[Result], "/tmp/test")