public static boolean checkKryoConfiguration ( JavaSparkContext javaSparkContext , Logger log ) { //Check if kryo configuration is correct: String serializer = javaSparkContext . getConf ( ) . get ( "spark.serializer" , null ) ; if ( serializer != null && serializer . equals ( "org.apache.spark.serializer.KryoSerializer" ) ) { String kryoRegistrator = javaSparkContext . getConf ( ) . get ( "spark.kryo.registrator" , null ) ; if ( kryoRegistrator == null || ! kryoRegistrator . equals ( "org.nd4j.Nd4jRegistrator" ) ) { //It's probably going to fail later due to Kryo failing on the INDArray deserialization (off-heap data) //But: the user might be using a custom Kryo registrator that can handle ND4J INDArrays, even if they // aren't using the official ND4J-provided one //Either way: Let's test serialization now of INDArrays now, and fail early if necessary SerializerInstance si ; ByteBuffer bb ; try { si = javaSparkContext . env ( ) . serializer ( ) . newInstance ( ) ; bb = si . serialize ( Nd4j . linspace ( 1 , 5 , 5 ) , null ) ; } catch ( Exception e ) { //Failed for some unknown reason during serialization - should never happen throw new RuntimeException ( KRYO_EXCEPTION_MSG , e ) ; } if ( bb == null ) { //Should probably never happen throw new RuntimeException ( KRYO_EXCEPTION_MSG + "\n(Got: null ByteBuffer from Spark SerializerInstance)" ) ; } else { //Could serialize successfully, but still may not be able to deserialize if kryo config is wrong boolean equals ; INDArray deserialized ; try { deserialized = ( INDArray ) si . deserialize ( bb , null ) ; //Equals method may fail on malformed INDArrays, hence should be within the try-catch equals = Nd4j . linspace ( 1 , 5 , 5 ) . equals ( deserialized ) ; } catch ( Exception e ) { throw new RuntimeException ( KRYO_EXCEPTION_MSG , e ) ; } if ( ! equals ) { throw new RuntimeException ( KRYO_EXCEPTION_MSG + "\n(Error during deserialization: test array" + " was not deserialized successfully)" ) ; } //Otherwise: serialization/deserialization was successful using Kryo return true ; } } } return true ; }
Check the spark configuration for incorrect Kryo configuration logging a warning message if necessary