Мигание потребляет ошибку сериализации киро файла s3 parquet - PullRequest
Новая
       38

Мигание потребляет ошибку сериализации киро файла s3 parquet

0 голосов
/ 21 января 2019

Мы хотим использовать паркетный файл из s3

Мой фрагмент кода такой.Мои входные файлы - это паркетные файлы, закодированные в protobufКласс protobuf - Pageview.class. 

import com.twitter.chill.protobuf.ProtobufSerializer;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.scala.hadoop.mapreduce.HadoopInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.parquet.proto.ProtoParquetInputFormat;
import org.apache.hadoop.fs.Path;
import scala.Tuple2;

public class ParquetReadJob {
    public static void main(String... args) throws Exception {

        ExecutionEnvironment ee = ExecutionEnvironment.getExecutionEnvironment();
        ee.getConfig().registerTypeWithKryoSerializer(StandardLog.Pageview.class, ProtobufSerializer.class);
        String path = args[0];

        Job job = Job.getInstance();
        job.setInputFormatClass(ProtoParquetInputFormat.class);

        HadoopInputFormat<Void, StandardLog.Pageview> hadoopIF =
                new HadoopInputFormat<> (new ProtoParquetInputFormat<>(), Void.class, StandardLog.Pageview.class, job);



        ProtoParquetInputFormat.addInputPath(job, new Path(path));
        DataSource<Tuple2<Void, StandardLog.Pageview>> dataSet = ee.createInput(hadoopIF).setParallelism(10);


        dataSet.print();
    }
}

Всегда есть ошибки: 

com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
Serialization trace:
supportCrtSize_ (access.Access$AdPositionInfo)
adPositionInfo_ (access.Access$AccessRequest)
accessRequest_ (com.adshonor.proto.StandardLog$Pageview$Builder)
    at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
    at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
    at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:730)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
    at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
    at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
    at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
    at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:730)
    at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)
    at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
    at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
    at com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:42)
    at com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:33)
    at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:315)
    at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:55)
    at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:106)
    at org.apache.flink.runtime.io.network.api.reader.AbstractRecordReader.getNextRecord(AbstractRecordReader.java:72)
    at org.apache.flink.runtime.io.network.api.reader.MutableRecordReader.next(MutableRecordReader.java:47)
    at org.apache.flink.runtime.operators.util.ReaderIterator.next(ReaderIterator.java:73)
    at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:216)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.UnsupportedOperationException
    at java.util.Collections$UnmodifiableCollection.add(Collections.java:1055)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)
    at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
    at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
    at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
    ... 23 more

Может кто-нибудь посоветовать мне, как написать программу пакетной обработки, которая может принимать файлы такого типа?

1 Ответ

0 голосов
/ 05 марта 2019

Я также столкнулся с этой проблемой. Я нашел this и this в ожидающем PR для flink-protobuf, который его решил.

Вам необходимо добавить классы NonLazyProtobufSerializer и ProtobufKryoSerializer в ваш проект. и зарегистрируйте NonLazyProtobufSerializer в качестве сериализатора Kryo по умолчанию для типа сообщения: 

env.getConfig().addDefaultKryoSerializer(Message.class, NonLazyProtobufSerializer.class);

От авторов JavaDocs:

Это обходной путь для проблемы, возникающей при использовании источника данных из Kafka в Flink TableEnvironment. Для полей, объявленных с типом 'string' в .proto, соответствующее поле на Класс Java объявил тип «Объект». Фактический тип этих полей на объектах, возвращаемых Message.parseFrom (byte []) является 'ByteArray'. Но методы getter для этих полей возвращают 'String', ленивая замена базового поля ByteArray на String, когда это необходимо.

Надеюсь, это поможет.

...