public void sink(Map<String, Object> settings, TupleEntry tupleEntry, OutputCollector outputCollector)
          throws IOException {

    String rowKeyField = SettingsHelper.getMappingRowKeyField(settings);

    Tuple key = tupleEntry.selectTuple(new Fields(rowKeyField));
    ByteBuffer keyBuffer = SerializerHelper.serialize(key.get(0));

    Map<String, String> dataTypes = SettingsHelper.getDynamicTypes(settings);
    Map<String, String> dynamicMappings = SettingsHelper.getDynamicMappings(settings);

    AbstractType columnNameType = SerializerHelper.inferType(dataTypes.get("columnName"));

    Object columnNameFieldSpec = dynamicMappings.get("columnName");
    List<String> columnNameFields = new ArrayList<String>();
    if (columnNameFieldSpec instanceof String) {
      columnNameFields.add((String) columnNameFieldSpec);
    } else {
      columnNameFields.addAll((List<String>) columnNameFieldSpec);
    }

    List tupleEntryColumnNameValues = new ArrayList();
    for (String columnNameField : columnNameFields) {
      try {
        Object columnNameValue = tupleEntry.getObject(columnNameField);
        logger.debug("column name component field: {}", columnNameField);
        logger.debug("column name component value: {}", columnNameValue);
        tupleEntryColumnNameValues.add(columnNameValue);
      } catch (FieldsResolverException e) {
        throw new RuntimeException("Couldn't resolve column name field: " + columnNameField);
      }
    }

    String columnValueField = (String) dynamicMappings.get("columnValue");
    Object tupleEntryColumnValueValue = null;
    if (columnValueField != null) {
      try {
        tupleEntryColumnValueValue = tupleEntry.getObject(columnValueField);
        logger.debug("column value field: {}", columnValueField);
        logger.debug("column value value: {}", tupleEntryColumnValueValue);
      } catch (FieldsResolverException e) {
        throw new RuntimeException("Couldn't resolve column value field: " + columnValueField);
      }
    }

    List<Mutation> mutations = new ArrayList<Mutation>();
    if (tupleEntryColumnNameValues.size() > 0) {

      ByteBuffer columnName = null;
      if (columnNameType instanceof CompositeType) {
        logger.debug("CompositeType");
        columnName = SerializerHelper.serializeComposite(tupleEntryColumnNameValues, (CompositeType) columnNameType);
      } else {
        logger.debug("SimpleType");
        columnName = SerializerHelper.serialize(tupleEntryColumnNameValues.get(0));
      }

      ByteBuffer serializedColValue;
      if (tupleEntryColumnValueValue != null) {
        serializedColValue = SerializerHelper.serialize(tupleEntryColumnValueValue);
      } else {
        byte[] emptyBytes = {};
        serializedColValue = ByteBuffer.wrap(emptyBytes);
      }

      Mutation mutation = Util.createColumnPutMutation(columnName, serializedColValue);
      mutations.add(mutation);
    }

    outputCollector.collect(keyBuffer, mutations);
  }