Iterator names = properties.keySet().iterator();
while (names.hasNext()) {
    // Identify the property name and value(s) to be assigned
    String name = (String) names.next();
    if (name == null) {
        continue;
    }
    Object value = properties.get(name);
    // Perform the assignment for this property
    setProperty(bean, name, value);
}

public <T extends Map<String,Object>, C> List<C> convertMapToBean(List<T> list,
  Class<C> clazz) {
    List<C> beanList = new ArrayList<C>();

    for (T item : list) {
        C bean = null;
        try {
            bean = clazz.newInstance();
            BeanUtils.populate(bean, item);
        } catch (InstantiationException e) {
            new IllegalArgumentException("Cannot initiate class",e);
        } catch (IllegalAccessException e) {
            new IllegalStateException("Cannot access the property",e);
        } catch (InvocationTargetException e) {
            new IllegalArgumentException(e);
        }
        beanList.add(bean);
    }
    return beanList;
}

public <T extends Map<String, Object>, C> List<C> convertMapToBean(
    List<T> list, Class<C> clazz) {
    List<C> beanList = new ArrayList<C>();

    for (T source : list) {
        C bean = null;

        try {
            bean = clazz.newInstance();

            PropertyDescriptor[] targetPds = PropertyUtils.getPropertyDescriptors(clazz);

            for (PropertyDescriptor desc : targetPds) {
                Object value = source.get(desc.getName());
                if (value != null) {
                    Method writeMethod = desc.getWriteMethod();
                    if (writeMethod != null) {
                        writeMethod.invoke(bean, new Object[] { value });
                    }
                }
            }
        } catch (InstantiationException e) {
            new IllegalArgumentException("Cannot initiate class",e);
        } catch (IllegalAccessException e) {
            new IllegalStateException("Cannot access the property",e);
        } catch (InvocationTargetException e) {
            new IllegalArgumentException(e);
        }
        beanList.add(bean);
    }
    return beanList;
}

public <T extends Map<String, Object>, C> List<C> convertMapToBean(List<T> list, Class<C> targetClass) {
    List<C> beanList = new ArrayList<C>();

    for (Map<String, Object> map : list) {
        C bean = null;
        try {
            bean = targetClass.newInstance();
            BeanUtils.setValues(bean, map, null);
        } catch (InstantiationException e) {
            new IllegalArgumentException("Cannot initiate class", e);
        } catch (IllegalAccessException e) {
            new IllegalStateException("Cannot access the property", e);
        }
        beanList.add(bean);
    }
    return beanList;
}

public <T extends Map<String, Object>, C> List<C> convertMapToBean(
    List<T> list, Class<C> clazz) {
    List<C> beanList = new ArrayList<C>();
    for (Map<String, Object> source : list) {
        C bean = toBean(source, clazz);
    beanList.add(bean);

    }
    return beanList;
}

private <C> C toBean(Map<String, Object> source, Class<C> targetClass) {
    C bean = null;
    try {
        bean = targetClass.newInstance();
        PropertyDescriptor[] targetPds = BeanUtils.getPropertyDescriptors(targetClass);

        for (PropertyDescriptor desc : targetPds) {
            Object value = source.get(desc.getName());
            if (value != null) {
                Method writeMethod = desc.getWriteMethod();
                if (writeMethod != null) {
                    writeMethod.invoke(bean, new Object[] { value });
                }
            }
        }
    } catch (InstantiationException e) {
        new IllegalArgumentException("Cannot initiate class",e);
    } catch (IllegalAccessException e) {
        new IllegalStateException("Cannot access the property",e);
    } catch (InvocationTargetException e) {
        new IllegalArgumentException(e);
    }
    return bean;
}

package org.apache.hadoop.mapred.lib.db;

import java.io.IOException;
import java.sql.SQLException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

public class OracleInputFormat<T extends DBWritable> extends DBInputFormat<T>{

    private DBConfiguration dbConf = null;
    private DBInputSplit split;

    @Override
    public RecordReader<LongWritable,T> getRecordReader(InputSplit split, JobConf job,
            Reporter reporter) throws IOException {
        dbConf = new DBConfiguration(job);
        this.split = (DBInputSplit)split;

        @SuppressWarnings("unchecked")
        Class inputClass = dbConf.getInputClass();
        try {
            @SuppressWarnings("unchecked")
            RecordReader<LongWritable,T> reader = new OracleRecordReader((DBInputSplit) split, inputClass, job);
            return reader;
        } catch (SQLException ex) {
            throw new IOException(ex.getMessage());
        }
    }

    public static void setInput(JobConf job, Class<? extends DBWritable> inputClass,
              String inputQuery, String inputCountQuery) {
        DBInputFormat.setInput(job, inputClass, inputQuery, inputCountQuery);
        job.setInputFormat(OracleInputFormat.class);
     }

    protected class OracleRecordReader extends DBRecordReader{
        protected OracleRecordReader(DBInputSplit split, Class<T> inputClass,
                JobConf conf) throws SQLException {
            super(split, inputClass, conf);
        }

        @Override
        protected String getSelectQuery() {
            long length = 0;
            long start = 0;
            try{
                length = split.getLength();
                start = split.getStart();
            } catch(IOException e){
                throw new IllegalArgumentException
                        ("cannot read length or start variable from DBInputSplit",e);
            }
            StringBuilder query = new StringBuilder();
            query.append(" SELECT * \r\n");
            query.append(" FROM (SELECT m.* , ROWNUM rno ");
            query.append("       FROM ( ");
            query.append(              dbConf.getInputQuery());
            query.append("             )  m");
            query.append("       WHERE ROWNUM <= " + start + " + " + length + ")");
            query.append(" WHERE RNO > " + start);
            System.out.println(query.toString());
            return query.toString();
        }
    }
}

package org.apache.hadoop.mapred.lib.db;
import org.apache.hadoop.mapred.JobConf;

public class OracleOutputFormat<K  extends DBWritable, V> extends DBOutputFormat<DBWritable, V>{
    @Override
    protected String constructQuery(String table, String[] fieldNames) {
            if(fieldNames == null) {
              throw new IllegalArgumentException("Field names may not be null");
            }
            StringBuilder query = new StringBuilder();
            query.append("INSERT INTO ").append(table);

            if (fieldNames.length > 0 && fieldNames[0] != null) {
              query.append(" (");
              for (int i = 0; i < fieldNames.length; i++) {
                query.append(fieldNames[i]);
                if (i != fieldNames.length - 1) {
                  query.append(",");
                }
              }
              query.append(")");
            }
            query.append(" VALUES (");

            for (int i = 0; i < fieldNames.length; i++) {
              query.append("?");
              if(i != fieldNames.length - 1) {
                query.append(",");
              }
            }
            query.append(")");
            return query.toString();
          }
    public static void setOutput(JobConf job, String tableName, String... fieldNames) {
        DBOutputFormat.setOutput(job, tableName, fieldNames);
        job.setOutputFormat(OracleOutputFormat.class);
    }
}

public class SampleJob {

    public static void main(String args[]) throws IOException, URISyntaxException{
        JobConf conf = new JobConf(SampleJob.class);
        initClasspath(conf);
        conf.setJobName("sampleJob");
        DBConfiguration.configureDB(conf, "oracle.jdbc.driver.OracleDriver",
                "jdbc:oracle:thin:@localhost:1525:TEST",
                "myuser", "mypassword");
        OracleInputFormat.setInput(conf, Query.class,
                "SELECT query, category, user_id FROM query_log ",
                "SELECT COUNT(*) FROM query_log");
        conf.setOutputKeyClass(Query.class);
        conf.setOutputValueClass(IntWritable.class);
        conf.setMapperClass(SampleMapper.class);
        conf.setReducerClass(SampleReducer.class);
        conf.setCombinerClass(SampleReducer.class);
        OracleOutputFormat.setOutput(conf, "category", "user_id","cnt");

        JobClient.runJob(conf);
    }

    private static void initClasspath(JobConf conf) throws URISyntaxException,
            IOException {
        DistributedCache.addCacheFile(new URI("lib/ojdbc5-11.1.0.6.jar"), conf);
        DistributedCache.addFileToClassPath(new Path("lib/ojdbc5-11.1.0.6.jar"), conf);
    }
}