@@ -0,0 +1,2369 @@
+package de.bigprak.transform.csv;%0A%0Aimport java.io.File;%0Aimport java.nio.charset.StandardCharsets;%0Aimport java.util.Arrays;%0A%0Aimport org.apache.commons.csv.CSVFormat;%0Aimport org.apache.commons.csv.QuoteMode;%0Aimport org.apache.commons.io.FileUtils;%0Aimport org.apache.flink.api.java.ExecutionEnvironment;%0Aimport org.apache.flink.api.java.tuple.Tuple3;%0A%0Apublic class CSVParsingTest %7B%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();%0A%0A        CSVFormat csvFormat = CSVFormat.newFormat(',')%0A                .withEscape('%5C%5C')%0A                .withQuoteMode(QuoteMode.NONE)%0A                .withNullString(%22%22);%0A%0A        File in = File.createTempFile(%22de.bigprak.transform.csv%22, %22CSVParsingTest-in%22);%0A        File out = File.createTempFile(%22de.bigprak.transform.csv%22, %22CSVParsingTest-out%22);%0A        in.deleteOnExit();%0A        out.deleteOnExit();%0A        out.delete();%0A%0A        // Extract (in this order) the id, the first Name and the message;%0A        // ignore second name and time%0A        String csv = %22Peter,Franz,1,Hallo Franz%5C%5C, wie geht es dir?,1456076785%5Cn%22 +%0A                     %22Franz,Peter,2,Sehr gut%5C%5C, danke%5C%5C, und dir?,1456076892%5Cn%22 +%0A                     %22Peter,Franz,3,Alles in Ordnung!,1456076899%22;%0A        int%5B%5D indices = %7B2,0,3%7D;%0A        Class%3C?%3E%5B%5D types = %7BLong.class, String.class, String.class%7D;%0A%0A        FileUtils.writeStringToFile(in, csv, StandardCharsets.UTF_8);%0A%0A        env.readFile(new CommonsCSVInputFormat(csvFormat), in.getAbsolutePath())%0A           .map(new CSVRecordToTupleMap%3CTuple3%3CLong, String, String%3E%3E(indices, types))%0A           // necessary, because flink cannot deduce return type from generic Mapper%0A           .returns(%22Tuple3%3CLong, String, String%3E%22)%0A           .writeAsCsv(out.getAbsolutePath());%0A%0A        env.setParallelism(1);%0A        env.execute();%0A%0A        String result = FileUtils.readFileToString(out, StandardCharsets.UTF_8);%0A%0A        // show results%0A        System.out.println();%0A        System.out.println(%22Input csv: %22);%0A        System.out.println(csv);%0A%0A        System.out.println();%0A        System.out.println(%22Extraction parameters: %22 + Arrays.toString(indices) + %22 (%22 + Arrays.toString(types) + %22)%22);%0A%0A        System.out.println();%0A        System.out.println(%22Resulting csv: %22);%0A        System.out.println(result);%0A    %7D%0A%7D%0A

Add test for the new CommonsCSVInputFormat and CSVRecordToTupleMap