Получение следующей ошибки при запуске моего лучевого трубопровода: < /p>
Java.lang.illegalargumentException: невозможно сериализовать dofnwitexecutionInformation {dofn = writewithappendTogile $ createTrailerDofn@57b711b6, mainoutputtag = tag, tag, tag, tag, tag, tag, tag, tag, tag, tag, ateTrailerDofn@57b711b6. schemainformation = dofnschemainformation {elementConverters = [], FieldAccessDescriptor =*}} < /p>
Причина: java.io.noterializableException: PipelineOptions Объекты не сериализируются и не должны были внедрены в преобразование (вы оценивали в почеле или в поле. Вместо этого, если вы используете DOFN, доступ к PipelineOptions во время выполнения через ProcessContext/startBundLecontext/FinishBundLecontext.getPipelineOptions () или предварительно экспрессировать необходимые поля из PipelineOptions в конструкцию Pipeline.private class CreateTrailerDoFn extends DoFn {
@ProcessElement
public void processElement(ProcessContext context) {
final int[] count = {0};
String timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new java.util.Date());
T data = context.element();
data.getContent().forEach(row -> {
LOG.info(String.format("data : %s", row));
count[0]++;
});
String trailerRow = String.format("%s,%s", count[0], timeStamp);
LOG.info(trailerRow);
context.output(trailerRow);
}
}
Подробнее здесь: https://stackoverflow.com/questions/741 ... of-pipelin
