поэтому я написал конвейер луча apache, который читает файл, который содержит 99 других файлов, вычисляет контрольную сумму и создает пару ключ-значение файла и его контрольную сумму, что мне нужно сделать, это записать эти пары ключ-значение в Файл manifest.json В настоящее время у меня возникают проблемы с сериализацией, и любые советы и помощь могут быть удивительными.
Вот мой код:
public class BeamPipeline {
private static final Logger log = LoggerFactory.getLogger(BeamPipeline.class);
public static interface MyOptions extends PipelineOptions {
@Description("Input Path(with gs:// prefix)")
String getInput();
void setInput(String value);
}
public static void main(String[] args) {
MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(MyOptions.class);
Pipeline p = Pipeline.create(options);
JsonObject obj = new JsonObject();
File dir = new File(options.getInput());
for (File file : dir.listFiles()) {
String inputString = file.toString();
p
.apply("Match Files", FileIO.match().filepattern(inputString))
.apply("Read Files", FileIO.readMatches())
.apply(MapElements.via(new SimpleFunction<FileIO.ReadableFile, KV<String, String>>() {
public KV<String, String> apply(FileIO.ReadableFile file) {
String temp = null;
try {
temp = file.readFullyAsUTF8String();
} catch (IOException e) {
}
String sha256hex = org.apache.commons.codec.digest.DigestUtils.sha256Hex(temp);
obj.addProperty(temp, sha256hex);
String json = obj.toString();
try (FileWriter fileWriter = new FileWriter("./manifest.json")) {
fileWriter.write(json);
} catch (IOException e) {
}
return KV.of(file.getMetadata().resourceId().toString(), sha256hex);
}
}))
.apply("Print", ParDo.of(new DoFn<KV<String, String>, Void>() {
@ProcessElement
public void processElement(ProcessContext c) {
log.info(String.format("File: %s, SHA-256 %s", c.element().getKey(), c.element().getValue()));
}
}));
}
p.run();
}
}
Вот мои ошибки в данный момент:
"main" java.lang.IllegalArgumentException: unable to serialize DoFnAndMainOutput{doFn=org.apache.beam.sdk.transforms.MapElements$1@50756c76, mainOutputTag=Tag<output>}
Caused by: java.io.NotSerializableException: com.google.gson.JsonObject