import org.apache.cassandra.db.Column; //导入依赖的package包/类
public void map(ByteBuffer key, SortedMap<ByteBuffer, Column> columns, Context context) throws IOException, InterruptedException
{
    for (Column column : columns.values())
    {
        String name  = ByteBufferUtil.string(column.name());
        String value = null;
        
        if (name.contains("int"))
            value = String.valueOf(ByteBufferUtil.toInt(column.value()));
        else
            value = ByteBufferUtil.string(column.value());
                       
        logger.debug("read {}:{}={} from {}",
                     new Object[] {ByteBufferUtil.string(key), name, value, context.getInputSplit()});

        StringTokenizer itr = new StringTokenizer(value);
        while (itr.hasMoreTokens())
        {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
private void add(ByteBuffer[] fullName, int idx, Column column)
{
    ByteBuffer columnName = fullName[idx];
    if (fullName.length == idx + 2)
    {
        // It's a collection
        Value v = map.get(columnName);
        if (v == null)
        {
            v = new Collection();
            map.put(columnName, v);
        }
        assert v instanceof Collection;

        ((Collection)v).add(Pair.create(fullName[idx + 1], column));
    }
    else
    {
        assert !map.containsKey(columnName);
        map.put(columnName, new Simple(column));
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
/** convert a column to a tuple */
protected Tuple columnToTuple(Column col, CfDef cfDef, AbstractType comparator) throws IOException
{
    Tuple pair = TupleFactory.getInstance().newTuple(2);

    // name
    if(comparator instanceof AbstractCompositeType)
        setTupleValue(pair, 0, composeComposite((AbstractCompositeType)comparator,col.name()));
    else
        setTupleValue(pair, 0, comparator.compose(col.name()));

    // value
    Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
    if (validators.get(col.name()) == null)
    {
        Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
        setTupleValue(pair, 1, marshallers.get(MarshallerType.DEFAULT_VALIDATOR).compose(col.value()));
    }
    else
        setTupleValue(pair, 1, validators.get(col.name()).compose(col.value()));
    return pair;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
@Override
public void nextColumn(Column column)
{
    // avoid adding already expired or deleted columns to the index
    if (column.isMarkedForDelete(start))
        return;

    ColumnIndex columnIndex = supportedIndexes.get(column.name());
    if (columnIndex == null)
        return;

    Index index = indexes.get(column.name());
    if (index == null)
        indexes.put(column.name(), (index = new Index(columnIndex)));

    index.add(column.value().duplicate(), currentKey, currentKeyPosition);
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public static ColumnFamily remoteStorageQuery(String target, String dataTag){
	try {
		List<ReadCommand> command = new ArrayList<ReadCommand>();
		command.add(new SliceFromReadCommand(
				Metadata.MetaData_KS,
				ByteBufferUtil.bytes(target), 
				new QueryPath(Metadata.MetadataLog_CF),
				Column.decomposeName(String.valueOf(0) , "", dataTag, "value"),
				Column.decomposeName(String.valueOf(Long.MAX_VALUE) , "", dataTag, "value"),
				false,
				Integer.MAX_VALUE));

		List<Row> rows = StorageProxy.read(command, ConsistencyLevel.ANY);
		return rows.get(0).cf;
		 
	} catch (Exception e) {
		return null;
	}
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
private ColumnFamily remoteStorageQuery(String target, String dataTag){
	try {
		List<ReadCommand> command = new ArrayList<ReadCommand>();
		command.add(new SliceFromReadCommand(
				Metadata.MetaData_KS,
				ByteBufferUtil.bytes(target), 
				new QueryPath(Metadata.MetadataRegistry_CF),
				Column.decomposeName(dataTag, "admin_tag"),
				Column.decomposeName(dataTag, "admin_tag"),
				false,
				Integer.MAX_VALUE));

		List<Row> rows = StorageProxy.read(command, ConsistencyLevel.ANY);
		return rows.get(0).cf;
		 
	} catch (Exception e) {
		return null;
	}
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
@Override protected void map(AegisthusKey key, AtomWritable value,
        Context context)
        throws IOException, InterruptedException {
    if (currentKey == null) {
        currentKey = key.getKey();
    } else if (!currentKey.equals(key.getKey())) {
        flushCgm(context);
        currentKey = key.getKey();
    }

    OnDiskAtom atom = value.getAtom();
    if (atom == null) {
        LOG.warn("Got null atom for key {}.", cfMetaData.getKeyValidator().compose(key.getKey()));
        return;
    }

    if (atom instanceof Column) {
        cgmBuilder.add((Column) atom);
    } else {
        LOG.error("Non-colum atom. {} {}", atom.getClass(), atom);
        throw new IllegalArgumentException("Got a non-column Atom.");
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public void finalizeReduce() {
    if (currentColumn != null) {
        columns.add(currentColumn);
    }

    // When cassandra compacts it removes columns that are in deleted rows
    // that are older than the deleted timestamp.
    // we will duplicate this behavior. If the etl needs this data at some
    // point we can change, but it is only available assuming
    // cassandra hasn't discarded it.
    Iterator<OnDiskAtom> columnIterator = columns.iterator();
    while (columnIterator.hasNext()) {
        OnDiskAtom atom = columnIterator.next();
        if (atom instanceof RangeTombstone) {
            columnIterator.remove();
        } else if (atom instanceof Column && ((Column) atom).timestamp() <= this.deletedAt) {
            columnIterator.remove();
        }
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public static Collection<Column> makeColumns(Message msg) throws IOException
{
    Collection<Column> cols = new ArrayList<Column>();
    DataInputStream in = new DataInputStream(new ByteArrayInputStream(msg.getMessageBody()));
    int count = in.readInt();
    for (int i = 0; i < count; i++)
    {
        byte[] name = new byte[in.readInt()];
        in.readFully(name);
        byte[] value = new byte[in.readInt()];
        in.readFully(value);
        cols.add(new Column(ByteBuffer.wrap(name), ByteBuffer.wrap(value)));
    }
    in.close();
    return cols;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public void map(ByteBuffer key, SortedMap<ByteBuffer, Column> columns, Context context) throws IOException, InterruptedException
{
    long sum = 0;
    for (Column column : columns.values())
    {
        logger.debug("read " + key + ":" + column.name() + " from " + context.getInputSplit());
        sum += ByteBufferUtil.toLong(column.value());
    }
    context.write(new Text(ByteBufferUtil.string(key)), new LongWritable(sum));
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
private static Mutation getMutation(Text word, int sum)
{
    org.apache.cassandra.thrift.Column c = new org.apache.cassandra.thrift.Column();
    c.setName(Arrays.copyOf(word.getBytes(), word.getLength()));
    c.setValue(ByteBufferUtil.bytes(sum));
    c.setTimestamp(System.currentTimeMillis());

    Mutation m = new Mutation();
    m.setColumn_or_supercolumn(new ColumnOrSuperColumn());
    m.column_or_supercolumn.setColumn(c);
    return m;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public void add(Column c)
{
    current.add(isDead(c) ? null : value(c));
    if (timestamps != null)
    {
        timestamps[current.size() - 1] = isDead(c) ? -1 : c.timestamp();
    }
    if (ttls != null)
    {
        int ttl = -1;
        if (!isDead(c) && c instanceof ExpiringColumn)
            ttl = c.getLocalDeletionTime() - (int) (now / 1000);
        ttls[current.size() - 1] = ttl;
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public Column getSimple(ByteBuffer key)
{
    Value v = map.get(key);
    if (v == null)
        return null;

    assert v instanceof Simple;
    return ((Simple)v).column;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public List<Pair<ByteBuffer, Column>> getCollection(ByteBuffer key)
{
    Value v = map.get(key);
    if (v == null)
        return null;

    assert v instanceof Collection;
    return (List<Pair<ByteBuffer, Column>>)v;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
protected List<Column> unthriftify(ColumnOrSuperColumn cosc)
{
    if (cosc.counter_column != null)
        return Collections.<Column>singletonList(unthriftifyCounter(cosc.counter_column));
    if (cosc.counter_super_column != null)
        return unthriftifySuperCounter(cosc.counter_super_column);
    if (cosc.super_column != null)
        return unthriftifySuper(cosc.super_column);
    assert cosc.column != null;
    return Collections.<Column>singletonList(unthriftifySimple(cosc.column));
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
private List<Column> unthriftifySuper(SuperColumn super_column)
{
    List<Column> columns = new ArrayList<Column>(super_column.columns.size());
    for (org.apache.cassandra.thrift.Column column : super_column.columns)
    {
        Column c = unthriftifySimple(column);
        columns.add(c.withUpdatedName(CompositeType.build(super_column.name, c.name())));
    }
    return columns;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
private List<Column> unthriftifySuperCounter(CounterSuperColumn super_column)
{
    List<Column> columns = new ArrayList<Column>(super_column.columns.size());
    for (CounterColumn column : super_column.columns)
    {
        Column c = unthriftifyCounter(column);
        columns.add(c.withUpdatedName(CompositeType.build(super_column.name, c.name())));
    }
    return columns;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
{
    maybeInit();
    if (rows == null)
        return endOfData();

    Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next = wideColumns.next();
    lastColumn = next.right.values().iterator().next().name().duplicate();

    maybeIncreaseRowCounter(next);
    return next;
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
/**
 * Increases the row counter only if we really moved to the next row.
 * @param next just fetched row slice
 */
private void maybeIncreaseRowCounter(Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next)
{
    ByteBuffer currentKey = next.left;
    if (!currentKey.equals(lastCountedKey))
    {
        totalRead++;
        lastCountedKey = currentKey;
    }
}
 

import org.apache.cassandra.db.Column; //导入依赖的package包/类
public boolean next(ByteBuffer key, SortedMap<ByteBuffer, Column> value) throws IOException
{
    if (this.nextKeyValue())
    {
        key.clear();
        key.put(this.getCurrentKey().duplicate());
        key.flip();

        value.clear();
        value.putAll(this.getCurrentValue());

        return true;
    }
    return false;
}