TopCarriersReducer extends Reducer<Text, TextArrayWritable, Text, Text> { void persist(String origin, TreeBidiMap<Double, String> topCarriers) { String cqlQuery = "INSERT INTO capstone.airport (code, top_carriers) VALUES (?,?)"; PreparedStatement preparedStatement = connect.prepare(cqlQuery); Object[] values = topCarriers.values().toArray(); CollectionUtils.reverseArray(values); BoundStatement boundStatement = preparedStatement.bind(origin, newArrayList(values)); connect.executeAsync(boundStatement); }  @Override void reduce(Text key, Iterable<TextArrayWritable> values, Context context); static Logger log; }
@Test public void testCassandraInsert() { TopCarriersReducer reducer = new TopCarriersReducer(); TreeBidiMap<Double, String> topCarriers = new TreeBidiMap<>(); topCarriers.put(1., "AW"); topCarriers.put(2., "LA"); topCarriers.put(2., "LA"); topCarriers.put(2., "LA"); topCarriers.put(2., "LA"); topCarriers.put(2., "LA"); topCarriers.put(2., "LA"); reducer.persist("NYC", topCarriers); }