@@ -0,0 +1,2429 @@
+package de.bigprak.transform.csv;%0A%0Aimport java.lang.reflect.Constructor;%0A%0Aimport org.apache.commons.csv.CSVRecord;%0Aimport org.apache.flink.api.common.functions.MapFunction;%0Aimport org.apache.flink.api.java.tuple.Tuple;%0A%0A/**%0A * Extract certain columns from the %7B@link CSVRecord%7D and store them in a %7B@link Tuple%7D.%0A *%0A * @param %3CT%3E One of %7B@link Tuple1%7D, %7B@link Tuple2%7D, ...%0A */%0Apublic class CSVRecordToTupleMap%3CT extends Tuple%3E implements MapFunction%3CCSVRecord, T%3E %7B%0A    private static final long serialVersionUID = 1L;%0A%0A    private int%5B%5D indices;%0A    private Class%3C?%3E%5B%5D types;%0A%0A    /**%0A     * Specify the columns to extract.%0A     *%0A     * All types must have an one-argument constructor that takes a %7B@link String%7D and%0A     * match the generic T when put into a Tuple of the correct size.%0A     *%0A     * Example:%0A     * %3Cpre%3E%3Ccode%3E%0A     * // assume T = Tuple3&lt;Long, String, Boolean&gt;%0A     * csv = %22Peter,22,1456076785,true,Hallo World%22;%0A     * indices = %7B2, 0, 3%7D;%0A     * types = %7BLong.class, String.class, Boolean.class%7D;%0A     * %3C/code%3E%3C/pre%3E%0A     *%0A     * @param indices Select which columns to extract (0-based)%0A     * @param types Specify the type of each column%0A     * @throws IllegalArgumentException When indices and types differ in length%0A     */%0A    public CSVRecordToTupleMap(int%5B%5D indices, Class%3C?%3E%5B%5D types) %7B%0A        if(indices.length != types.length) %7B%0A            throw new IllegalArgumentException(%22indices and types have to be of same length%22);%0A        %7D%0A%0A        this.indices = indices;%0A        this.types = types;%0A    %7D%0A%0A    @Override%0A    @SuppressWarnings(%22unchecked%22)%0A    public T map(CSVRecord record) throws Exception %7B%0A        Tuple tuple = Tuple.getTupleClass(indices.length).newInstance();%0A%0A        for(int i = 0; i %3C indices.length; ++i) %7B%0A            if(indices%5Bi%5D %3E= record.size()) %7B%0A                throw new RuntimeException(%22Requested column %22 + indices%5Bi%5D + %22 but record only has %22 + record.size() + %22columns%22);%0A            %7D%0A%0A            Object value = null;%0A%0A            try %7B%0A                Constructor%3C?%3E constructor = types%5Bi%5D.getConstructor(String.class);%0A                value = constructor.newInstance(record.get(indices%5Bi%5D));%0A            %7D catch(Exception e) %7B%0A                throw new RuntimeException(%22Could not convert column %22 + indices%5Bi%5D + %22 to type %22 + types%5Bi%5D.getName(), e);%0A            %7D%0A%0A            tuple.setField(value, i);%0A        %7D%0A%0A        return (T) tuple;%0A    %7D%0A%7D%0A

Add mapper to convert CSVRecords to Tuples