/*
 * Copyright 2017 Philip Fritzsche <p-f@users.noreply.github.com>.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.gradoop.examples.io.mag.parse.flink;

import java.util.logging.Level;
import java.util.logging.Logger;
import one.p_f.testing.magimport.InputSchema;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.fs.Path;

/**
 * Main class for importing mag data via flink.
 */
public class ImportMain {

    /**
     * Logger of this class.
     */
    private static final Logger LOG
            = Logger.getLogger(ImportMain.class.getName());

    /**
     * @param args Usage: -i INPATH -o OUTPATH
     */
    public static void main(String[] args) {
        Options cliOptions = new Options();
        cliOptions.addOption("h", "help", false, "Show this help.");
        cliOptions.addRequiredOption("i", "input", true, "Input path.");
        cliOptions.addRequiredOption("o", "output", true, "Output path.");
        CommandLine cliConfig;
        try {
            cliConfig = new DefaultParser().parse(cliOptions, args);
        } catch (ParseException ex) {
            LOG.log(Level.SEVERE, "Failed to parse command line options: {0}",
                    ex.getMessage());
            showHelp(cliOptions);
            return;
        }
        if (cliConfig.hasOption('h')) {
            showHelp(cliOptions);
            return;
        }
        String inPath = cliConfig.getOptionValue('i');
        Path outPath = new Path(cliConfig.getOptionValue('o'));
        ExecutionEnvironment localEnv = ExecutionEnvironment
                .createLocalEnvironment();
        FlinkParser parser
                = new FlinkParser(inPath, localEnv, InputSchema.get());
        parser.call();

    }

    private static void showHelp(Options o) {
        new HelpFormatter().printHelp("java ImportMain", o, true);
    }

}

@@ -676,16 +676,44 @@
 flink;%0A%0A
+import java.io.IOException;%0A
 import j
@@ -1029,24 +1029,66 @@
 eException;%0A
+import org.apache.flink.api.java.DataSet;%0A
 import org.a
@@ -1167,16 +1167,428 @@
 fs.Path;
+%0Aimport org.gradoop.flink.io.api.DataSink;%0Aimport org.gradoop.flink.io.api.DataSource;%0Aimport org.gradoop.flink.io.impl.graph.GraphDataSource;%0Aimport org.gradoop.flink.io.impl.graph.tuples.ImportEdge;%0Aimport org.gradoop.flink.io.impl.graph.tuples.ImportVertex;%0Aimport org.gradoop.flink.io.impl.json.JSONDataSink;%0Aimport org.gradoop.flink.model.impl.LogicalGraph;%0Aimport org.gradoop.flink.util.GradoopFlinkConfig;
 %0A%0A/**%0A *
@@ -2946,30 +2946,1065 @@
     
-parser.call();%0A%0A    %7D%0A
+DataSet%3CImportVertex%3CString%3E%3E vertices;%0A        DataSet%3CImportEdge%3CString%3E%3E edges;%0A        try %7B%0A            vertices = parser.getVertices();%0A            edges = parser.getEdges();%0A        %7D catch (MagParserException mpe) %7B%0A            LOG.log(Level.SEVERE, %22Parsing failed.%22, mpe);%0A            return;%0A        %7D%0A        GradoopFlinkConfig config = GradoopFlinkConfig.createConfig(localEnv);%0A        DataSource source = new GraphDataSource(vertices, edges, config);%0A        LogicalGraph graph;%0A        try %7B%0A            graph = source.getLogicalGraph();%0A        %7D catch (IOException ioe) %7B%0A            LOG.log(Level.SEVERE, %22Failed to create logical graph.%22, ioe);%0A            return;%0A        %7D%0A        DataSink sink = new JSONDataSink(outPath.toString(), config);%0A        try %7B%0A            sink.write(graph, true);%0A        %7D catch (IOException ex) %7B%0A            LOG.log(Level.SEVERE, %22Failed to write graph.%22, ex);%0A            return;%0A        %7D%0A    %7D%0A%0A    /**%0A     * Helper method showing help.%0A     *%0A     * @param o Options to generate help from.%0A     */
 %0A   

Implement main.