Received an event in channel 0 while still having data from a record

[Newport, Billy]

Anyone seen this before:

 

Caused by: java.io.IOException: Received an event in channel 0 while still having data from a record. This indicates broken serialization logic. If you are using custom serialization code (Writable or Value types), check their serialization routines. In the case of Kryo, check the respective Kryo serializer.

     at org.apache.flink.runtime.io.network.api.reader.AbstractRecordReader.getNextRecord(AbstractRecordReader.java:98)

     at org.apache.flink.runtime.io.network.api.reader.MutableRecordReader.next(MutableRecordReader.java:42)

     at org.apache.flink.runtime.operators.util.ReaderIterator.next(ReaderIterator.java:73)

     at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:190)

     at org.apache.flink.runtime.taskmanager.Task.run(Task.java:642)

     at java.lang.Thread.run(Thread.java:745)

 

 

We’re on 1.1.4 right now. We’re reading parquet file using code like this:

 

           AvroParquetInputFormat<GenericRecord> inputFormat = new AvroParquetInputFormat<GenericRecord>();

           AvroParquetInputFormat.setAvroReadSchema(job, getMergeSchema(storeName, datasetName));

 

           // Get patch of input parquet file

           DatasetHdfsInfo info = getLastCompleteMergedDatasetHDFSInfo(storeName, datasetName);

          

           Path path = new Path(info.getRootDir());

 

           DataSet<Tuple2<Void, GenericRecord>> d = getExecutionEnvironment().readHadoopFile(inputFormat, Void.class, GenericRecord.class, path.toString(), job);

 

 

 

[M. Dale]

How were the Parquet files you are trying to read generated? Same version of libraries? I am successfully using the following Scala code to read Parquet files using the HadoopInputFormat wrapper. Maybe try that in Java?

val hadoopInputFormat =
  new HadoopInputFormat[Void, GenericRecord](new AvroParquetInputFormat, classOf[Void], classOf[GenericRecord], job)

AvroParquetInputFormat.setAvroReadSchema(job, EventOnlyRecord.getClassSchema)

//APIF extends ParquetInputFormat which extends FileInputFormat (FIP)
//addInputPath is a static method on FIP.
val inputPath = new Path(input)
FileInputFormat.addInputPath(job, inputPath)

val rawEvents: DataSet[(Void, GenericRecord)] = env.createInput(hadoopInputFormat)

On 01/11/2017 03:16 PM, Newport, Billy wrote:

Anyone seen this before:

 

Caused by: java.io.IOException: Received an event in channel 0 while still having data from a record. This indicates broken serialization logic. If you are using custom serialization code (Writable or Value types), check their serialization routines. In the case of Kryo, check the respective Kryo serializer.

     at org.apache.flink.runtime.io.network.api.reader.AbstractRecordReader.getNextRecord(AbstractRecordReader.java:98)

     at org.apache.flink.runtime.io.network.api.reader.MutableRecordReader.next(MutableRecordReader.java:42)

     at org.apache.flink.runtime.operators.util.ReaderIterator.next(ReaderIterator.java:73)

     at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:190)

     at org.apache.flink.runtime.taskmanager.Task.run(Task.java:642)

     at java.lang.Thread.run(Thread.java:745)

 

 

We’re on 1.1.4 right now. We’re reading parquet file using code like this:

 

           AvroParquetInputFormat<GenericRecord> inputFormat = new AvroParquetInputFormat<GenericRecord>();

           AvroParquetInputFormat.setAvroReadSchema(job, getMergeSchema(storeName, datasetName));

 

           // Get patch of input parquet file

           DatasetHdfsInfo info = getLastCompleteMergedDatasetHDFSInfo(storeName, datasetName);

          

           Path path = new Path(info.getRootDir());

 

           DataSet<Tuple2<Void, GenericRecord>> d = getExecutionEnvironment().readHadoopFile(inputFormat, Void.class, GenericRecord.class, path.toString(), job);

 

 

 

[Aljoscha Krettek]

Hi Billy,

the stack trace seems to indicate that there is a problem at the point where the data sink is trying to read the input elements so it doesn't seem to be related to the source. Could you also post what sinks you have and what the type of the input elements of these sinks are?

Cheers,

Aljoscha

On Thu, 12 Jan 2017 at 04:45 M. Dale <[hidden email]> wrote:

How were the Parquet files you are trying to read generated? Same version of libraries? I am successfully using the following Scala code to read Parquet files using the HadoopInputFormat wrapper. Maybe try that in Java?

val hadoopInputFormat =
  new HadoopInputFormat[Void, GenericRecord](new AvroParquetInputFormat, classOf[Void], classOf[GenericRecord], job)

AvroParquetInputFormat.setAvroReadSchema(job, EventOnlyRecord.getClassSchema)

//APIF extends ParquetInputFormat which extends FileInputFormat (FIP)
//addInputPath is a static method on FIP.
val inputPath = new Path(input)
FileInputFormat.addInputPath(job, inputPath)

val rawEvents: DataSet[(Void, GenericRecord)] = env.createInput(hadoopInputFormat)

On 01/11/2017 03:16 PM, Newport, Billy wrote:

Anyone seen this before:

 

Caused by: java.io.IOException: Received an event in channel 0 while still having data from a record. This indicates broken serialization logic. If you are using custom serialization code (Writable or Value types), check their serialization routines. In the case of Kryo, check the respective Kryo serializer.

     at org.apache.flink.runtime.io.network.api.reader.AbstractRecordReader.getNextRecord(AbstractRecordReader.java:98)

     at org.apache.flink.runtime.io.network.api.reader.MutableRecordReader.next(MutableRecordReader.java:42)

     at org.apache.flink.runtime.operators.util.ReaderIterator.next(ReaderIterator.java:73)

     at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:190)

     at org.apache.flink.runtime.taskmanager.Task.run(Task.java:642)

     at java.lang.Thread.run(Thread.java:745)

 

 

We’re on 1.1.4 right now. We’re reading parquet file using code like this:

 

           AvroParquetInputFormat<GenericRecord> inputFormat = new AvroParquetInputFormat<GenericRecord>();

           AvroParquetInputFormat.setAvroReadSchema(job, getMergeSchema(storeName, datasetName));

 

           // Get patch of input parquet file

           DatasetHdfsInfo info = getLastCompleteMergedDatasetHDFSInfo(storeName, datasetName);

          

           Path path = new Path(info.getRootDir());

 

           DataSet<Tuple2<Void, GenericRecord>> d = getExecutionEnvironment().readHadoopFile(inputFormat, Void.class, GenericRecord.class, path.toString(), job);