So what should I do?
On Fri, Aug 7, 2015 at 12:01 PM, Fabian Hueske <fhue...@gmail.com> wrote:
> Ah, I checked the code.
>
> The BinaryInputFormat expects metadata which is written be the
> BinaryOutputFormat.
> So you cannot use the BinaryInputFormat to read a file which does not
> provide the metadata.
>
> 2015-08-07 11:53 GMT+02:00 Flavio Pompermaier <pomperma...@okkam.it>:
>
>> The file containing the serialized object is 7 bytes
>>
>> On Fri, Aug 7, 2015 at 11:49 AM, Fabian Hueske <fhue...@gmail.com> wrote:
>>
>>> This might be an issue with the blockSize parameter of the
>>> BinaryInputFormat.
>>> How large is the file with the single object?
>>>
>>> 2015-08-07 11:37 GMT+02:00 Flavio Pompermaier <pomperma...@okkam.it>:
>>>
>>>> I also tried with
>>>>
>>>> DataSet<RowBundle> ds = env.createInput(inputFormat).setParallelism(1);
>>>>
>>>> but I get the same error :(
>>>>
>>>> Moreover, in this example I put exactly one object per file so it
>>>> should be able to deserialize it, right?
>>>>
>>>> On Fri, Aug 7, 2015 at 11:33 AM, Fabian Hueske <fhue...@gmail.com>
>>>> wrote:
>>>>
>>>>> If you create your file by just sequentially writing all objects to
>>>>> the file using Kryo, you can only read it with a parallelism of 1.
>>>>> Writing binary files in a way that they can be read in parallel is a
>>>>> bit tricky (and not specific to Flink).
>>>>>
>>>>> 2015-08-07 11:28 GMT+02:00 Flavio Pompermaier <pomperma...@okkam.it>:
>>>>>
>>>>>> Hi to all,
>>>>>> I;m trying to read a file serialized with kryo but I get this
>>>>>> exception (due to the fact that the createInputSplits creates 8
>>>>>> inputsplits, where just one is not empty..).
>>>>>>
>>>>>> Caused by: java.io.IOException: Invalid argument
>>>>>> at sun.nio.ch.FileChannelImpl.position0(Native Method)
>>>>>> at sun.nio.ch.FileChannelImpl.position(FileChannelImpl.java:285)
>>>>>> at
>>>>>> org.apache.flink.core.fs.local.LocalDataInputStream.seek(LocalDataInputStream.java:57)
>>>>>> at
>>>>>> org.apache.flink.api.common.io.BinaryInputFormat.open(BinaryInputFormat.java:257)
>>>>>> at
>>>>>> org.apache.flink.api.common.io.BinaryInputFormat.open(BinaryInputFormat.java:46)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:151)
>>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>>>
>>>>>> -----------------------------------------------
>>>>>> My program is basically the following:
>>>>>>
>>>>>> public static void main(String[] args) throws Exception {
>>>>>>
>>>>>> ...
>>>>>> //try-with-resources used to autoclose resources
>>>>>> try (Output output = new Output(new
>>>>>> FileOutputStream("/tmp/KryoTest.ser"))) {
>>>>>> //serialise object
>>>>>> Kryo kryo=new Kryo();
>>>>>> kryo.writeClassAndObject(output, myObj);
>>>>>> } catch (FileNotFoundException ex) {
>>>>>> LOG.error(ex.getMessage(), ex);
>>>>>> }
>>>>>>
>>>>>> //deserialise object
>>>>>>
>>>>>> myObj=null;
>>>>>>
>>>>>> try (Input input = new Input( new
>>>>>> FileInputStream("/tmp/KryoTest.ser"))){
>>>>>> Kryo kryo=new Kryo();
>>>>>> myObj =(MyClass)kryo.readClassAndObject(input);
>>>>>> } catch (FileNotFoundException ex) {
>>>>>> LOG.error(ex.getMessage(), ex);
>>>>>> }
>>>>>>
>>>>>>
>>>>>> final ExecutionEnvironment env =
>>>>>> ExecutionEnvironment.getExecutionEnvironment();
>>>>>> env.registerTypeWithKryoSerializer(MyClass.class,
>>>>>> MyClassSerializer.class);
>>>>>> Configuration configuration = new Configuration();
>>>>>> configuration.setLong(BinaryInputFormat.BLOCK_SIZE_PARAMETER_KEY,
>>>>>> 64*1024*1024);
>>>>>>
>>>>>> TypeInformation<MyClass> typeInfo = new
>>>>>> GenericTypeInfo<>(MyClass.class);
>>>>>> final BinaryInputFormat<MyClass> inputFormat = new
>>>>>> TypeSerializerInputFormat<>(typeInfo);
>>>>>> inputFormat.setFilePath("file:/tmp/KryoTest.ser");
>>>>>> inputFormat.configure(configuration);
>>>>>>
>>>>>> DataSet<MyClass> ds = env.createInput(inputFormat);
>>>>>> ds.print();
>>>>>>
>>>>>> }
>>>>>>
>>>>>> private static final class MyClassSerializer extends
>>>>>> Serializer<MyClass> {
>>>>>>
>>>>>> @Override
>>>>>> public void write(Kryo kryo, Output output, MyClass object) {
>>>>>> kryo.writeClassAndObject(output, object);
>>>>>> }
>>>>>>
>>>>>> @Override
>>>>>> public MyClass read(Kryo kryo, Input input, Class<MyClass> type) {
>>>>>> return (MyClass) kryo.readClassAndObject(input);
>>>>>> }
>>>>>> }
>>>>>>
>>>>>> Am I doing something wrong?
>>>>>>
>>>>>> Best,
>>>>>> Flavio
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>>
>
