Check if generic Tuple serialization can be moved from Kryo to Cascading's own serialization

[fhueske]

Right now, Cascading on Flink uses Kryo to serialize tuples with unknown schema.
Wrapping to Cascading's own serialization logic in Flink's serializer might improve the performance.

[kkrugler]

I think it's likely this would significantly improve performance. Currently I see a lot of processor time being spent in threads that look like this:

"MapPartition (map-DB4889E9084A472E82A7F42B3482D1F6) (3/20)" daemon prio=10 tid=0x00007f556747d800 nid=0x6690 runnable [0x00007f5569647000]
   java.lang.Thread.State: RUNNABLE
    at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.write(DefaultSerializers.java:173)
    at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.write(DefaultSerializers.java:166)
    at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:199)
    at com.dataartisans.flink.cascading.types.tuple.DefinedTupleSerializer.serialize(DefinedTupleSerializer.java:142)
    at com.dataartisans.flink.cascading.types.tuple.DefinedTupleSerializer.serialize(DefinedTupleSerializer.java:29)
    at org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:56)
    at org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:79)
    at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:84)
    - locked <0x00000006e78d9510> (a org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer)
    at org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65)
    at com.dataartisans.flink.cascading.runtime.boundaryStages.BoundaryOutStage.receive(BoundaryOutStage.java:44)
    at com.dataartisans.flink.cascading.runtime.boundaryStages.BoundaryOutStage.receive(BoundaryOutStage.java:29)
    at cascading.flow.stream.element.FunctionEachStage$1.collect(FunctionEachStage.java:81)

[kkrugler]

Would it be OK to have a dependency on Hadoop for this? I assume yes, in which case it seems possible to wire up Cascading's TupleSerialization (or a version thereof) versus using Kryo.

[fhueske]

Flink has a dependency on Hadoop. So there should already be a transitive dependency on Hadoop.
If it is an option for you, you can also add types to the Fields declaration. In that case, the connector will use Flink's "native" serializers and comparators which support comparisons on the binary representation.

In any case, it would be good to improve the serialization for untyped records.