package org.tuberlin.de.read_data;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.text.DateFormat;
import java.text.SimpleDateFormat;

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.
 */

/**
 * Skeleton for a Flink Job.
 * <p/>
 * For a full example of a Flink Job, see the WordCountJob.java file in the
 * same package/directory or have a look at the website.
 * <p/>
 * You can also generate a .jar file that you can submit on your Flink
 * cluster.
 * Just type
 * mvn clean package
 * in the projects root directory.
 * You will find the jar in
 * target/flink-quickstart-0.1-SNAPSHOT-Sample.jar
 */
public class Job {

    public static void main(String[] args) throws Exception {
        // set up the execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<String> textInput = env.readTextFile(args[1]);
        DataSet<Taxidrive> taxidriveDataSet = textInput.flatMap(new TaxidriveReader());

        /**
         * Here, you can start creating your execution plan for Flink.
         *
         * Start with getting some data from the environment, like
         * 	env.readTextFile(textPath);
         *
         * then, transform the resulting DataSet<String> using operations
         * like
         * 	.filter()
         * 	.flatMap()
         * 	.join()
         * 	.coGroup()
         * and many more.
         * Have a look at the programming guide for the Java API:
         *
         * http://flink.apache.org/docs/latest/programming_guide.html
         *
         * and the examples
         *
         * http://flink.apache.org/docs/latest/examples.html
         *
         */

        // execute program
        env.execute("Flink Java API Skeleton");
    }


    public static final class TaxidriveReader implements FlatMapFunction<String, Taxidrive> {

        @Override
        public void flatMap(String value, Collector<Taxidrive> out)
                throws Exception {
            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Taxidrive taxidrive = new Taxidrive();

            String[] splittedText = value.split(",");
            taxidrive.setTaxiID(splittedText[0]);
            taxidrive.setLicenseID(splittedText[1]);
            taxidrive.setPickup_datetime(splittedText[2]);
            taxidrive.setDropoff_datetime(splittedText[3]);
            taxidrive.setTrip_time_in_secs(Integer.parseInt(splittedText[4]));
            taxidrive.setTrip_distance(Double.parseDouble(splittedText[5]));
            taxidrive.setPickup_longitude(Double.parseDouble(splittedText[6]));
            taxidrive.setPickup_latitude(Double.parseDouble(splittedText[7]));
            if (!splittedText[8].equals("")) taxidrive.setDropoff_longitude(Double.parseDouble(splittedText[8]));
            if (!splittedText[9].equals("")) taxidrive.setDropoff_latitude(Double.parseDouble(splittedText[9]));
            taxidrive.setPayment_type(splittedText[10]);
            taxidrive.setFare_amount(Double.parseDouble(splittedText[11]));
            taxidrive.setSurcharge(Double.parseDouble(splittedText[12]));
            taxidrive.setMta_tax(Double.parseDouble(splittedText[13]));
            taxidrive.setTip_amount(Double.parseDouble(splittedText[14]));
            taxidrive.setTolls_amount(Double.parseDouble(splittedText[15]));
            taxidrive.setTotal_amount(Double.parseDouble(splittedText[16]));
            out.collect(taxidrive);
        }
    }
}

@@ -1547,1138 +1547,8 @@
 ob %7B
-%0A%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        // set up the execution environment%0A        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();%0A%0A        DataSet%3CString%3E textInput = env.readTextFile(args%5B1%5D);%0A        DataSet%3CTaxidrive%3E taxidriveDataSet = textInput.flatMap(new TaxidriveReader());%0A%0A        /**%0A         * Here, you can start creating your execution plan for Flink.%0A         *%0A         * Start with getting some data from the environment, like%0A         * %09env.readTextFile(textPath);%0A         *%0A         * then, transform the resulting DataSet%3CString%3E using operations%0A         * like%0A         * %09.filter()%0A         * %09.flatMap()%0A         * %09.join()%0A         * %09.coGroup()%0A         * and many more.%0A         * Have a look at the programming guide for the Java API:%0A         *%0A         * http://flink.apache.org/docs/latest/programming_guide.html%0A         *%0A         * and the examples%0A         *%0A         * http://flink.apache.org/docs/latest/examples.html%0A         *%0A         */%0A%0A        // execute program%0A        env.execute(%22Flink Java API Skeleton%22);%0A    %7D%0A%0A
 %0A   
@@ -1638,17 +1638,16 @@
 rive%3E %7B%0A
-%0A
         
@@ -1727,24 +1727,8 @@
 out)
-%0A               
  thr
@@ -1747,89 +1747,8 @@
 n %7B%0A
-            DateFormat dateFormat = new SimpleDateFormat(%22yyyy-MM-dd HH:mm:ss%22);%0A
     

Remove unused skeleton code