Как мне сериализовать и записать JSON в файл? - PullRequest
Новая
       33

Как мне сериализовать и записать JSON в файл?

0 голосов
/ 24 апреля 2019

поэтому я написал конвейер луча 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

1 Ответ

1 голос
/ 24 апреля 2019

DoFns сериализуются со всеми объектами, доступ к которым осуществляется из Dofn. JsonObject не сериализуем. Они созданы из DoFn и переданы в DoFn, что делает DoFn не сериализуемым.

Вы можете создать JsonObject с помощью DoFn, чтобы избежать проблем с сериализацией. 

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);
        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);

                                JsonObject obj = new JsonObject();
                                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();
        }
}
...