import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
private Mutation parseLine(FileMapping mapping, String[] line) {
  RowMapping rowMapping = mapping.getRowMapping();
  // Construct the Mutation
  Mutation mutation = rowMapping.getRowId(line);
  int rowOffset = rowMapping.getLogicalOffset();
  assert null != mutation;

  // Build the Mutation - each "column" in the line of data
  for (int logicalOffset = 0; logicalOffset < mapping.numMappings(); logicalOffset++) {
    if (logicalOffset == rowOffset) {
      continue;
    }
    // Avoid calling getColumnMapping for the rowId offset
    ColumnMapping colMapping = mapping.getColumnMapping(logicalOffset);
    String value = line[logicalOffset];
    if (null == value) {
      LOG.debug("Saw null value at column offset {}", logicalOffset);
      continue;
    }
    colMapping.addColumns(mutation, value);
  }

  return mutation;
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
/**
 * Processes a record by extracting its field map and converting
 * it into a list of Mutations into Accumulo.
 */
public void accept(FieldMappable record)
    throws IOException, ProcessingException {
  Map<String, Object> fields = record.getFieldMap();

  Iterable<Mutation> putList = mutationTransformer.getMutations(fields);
  if (null != putList) {
    for (Mutation m : putList) {
      try {
        this.table.addMutation(m);
      } catch (MutationsRejectedException ex) {
        throw new IOException("Mutation rejected" , ex);
      }
    }
  }
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
/**
 * Write random entries.
 * <p>
 * Closes the writer after entries are written.
 *
 * @param writer
 *          Writer to write entries to.
 */
void writeRandomEntries(BatchWriter writer) throws MutationsRejectedException {
  for (int i = 0; i < rowCount; i++) {
    byte[] row = getRandomBytes(keyFieldSize, true);

    for (int j = 0; j < columnCount; j++) {
      byte[] colF = getRandomBytes(keyFieldSize, true);
      byte[] colQ = getRandomBytes(keyFieldSize, true);
      byte[] value = getRandomBytes(valueFieldSize, false);

      Mutation mutation = new Mutation(row);
      mutation.put(colF, colQ, VISIBILITY, value);
      writer.addMutation(mutation);
    }
  }

  writer.close();
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
/**
 * Get random mutations to be written.
 *
 * @return Mutations.
 */
List<Mutation> getMutations() {
  List<Mutation> mutations = new ArrayList<>(rowCount);
  for (int i = 0; i < rowCount; i++) {
    byte[] row = getRandomBytes(keyFieldSize, true);

    Mutation mutation = new Mutation(row);
    mutations.add(mutation);

    for (int j = 0; j < columnCount; j++) {
      byte[] colF = getRandomBytes(keyFieldSize, true);
      byte[] colQ = getRandomBytes(keyFieldSize, true);
      byte[] value = getRandomBytes(valueFieldSize, false);

      mutation.put(colF, colQ, VISIBILITY, value);
    }
  }
  return mutations;
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
/**
 * Writes the given data to Accumulo. The full combinatorial of values is written.
 *
 * @param rows
 *          Rows to write.
 * @param colFs
 *          Column families to write.
 * @param colQs
 *          Column qualifiers to write.
 * @param colVs
 *          Column visibilities to write.
 * @param values
 *          Values to write.
 */
private static void writeData(BatchWriter writer, Iterable<String> rows, Iterable<String> colFs, Iterable<String> colQs, Iterable<String> colVs,
    Iterable<String> values) throws MutationsRejectedException {
  List<Mutation> mutations = new ArrayList<>();

  for (String row : rows) {
    Mutation mutation = new Mutation(row);
    mutations.add(mutation);

    for (String colF : colFs) {
      for (String colQ : colQs) {
        for (String colV : colVs) {
          for (String value : values) {
            mutation.put(colF, colQ, new ColumnVisibility(colV), value);
          }
        }
      }
    }
  }

  writer.addMutations(mutations);
  writer.flush();
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Test
public void addMutationsTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);

  BatchWriter writer = new SignedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("config1.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));

  List<Mutation> mutations = new ArrayList<>();
  Mutation mutation = new Mutation("row");
  mutation.put("colF", "colQ", "val");

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(1)).addMutation(any()); // 1 time
  verify(mockSignatureWriter, never()).addMutation(any());

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(3)).addMutation(any()); // 1 + 2 times
  verify(mockSignatureWriter, never()).addMutation(any());

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(6)).addMutation(any()); // 1 + 2 + 3 times
  verify(mockSignatureWriter, never()).addMutation(any());
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Test
public void addMutationsExternalTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);
  when(mockConnector.createBatchWriter(SIG_TABLE, null)).thenReturn(mockSignatureWriter);

  BatchWriter writer = new SignedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("config3.ini"), aliceKeyContainers.get(ValueSigner.ECDSA));

  List<Mutation> mutations = new ArrayList<>();
  Mutation mutation = new Mutation("row");
  mutation.put("colF", "colQ", "val");

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(1)).addMutation(any()); // 1 time
  verify(mockSignatureWriter, times(1)).addMutation(any()); // 1 time

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(3)).addMutation(any()); // 1 + 2 times
  verify(mockSignatureWriter, times(3)).addMutation(any()); // 1 + 2 times

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(6)).addMutation(any()); // 1 + 2 + 3 times
  verify(mockSignatureWriter, times(6)).addMutation(any()); // 1 + 2 + 3 times
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Test
public void addMutationsTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);
  BatchWriter writer = new EncryptedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("encrypt-value.ini"), KEYS);

  List<Mutation> mutations = new ArrayList<>();
  Mutation mutation = new Mutation("row");
  mutation.put("colF", "colQ", "val");

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(1)).addMutation(any()); // 1 time

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(3)).addMutation(any()); // 1 + 2 times

  mutations.add(mutation);
  writer.addMutations(mutations);
  verify(mockWriter, times(6)).addMutation(any()); // 1 + 2 + 3 times
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
public void put(final String tableName, final String rowId, final Collection<PutMutation> columns) throws IOException {

    BatchWriter writer = batchWriter(10000000L, tableName);

    Mutation mutation = new Mutation(new Text(rowId));
    for (final PutMutation column : columns) {
        mutation.put(new Text(column.getColumnFamily()), new Text(column.getColumnQualifier()),
                new ColumnVisibility(column.getColumnVisibilty()), new Value(column.getBuffer().getBytes()));
    }

    try {
        writer.addMutation(mutation);
        writer.close();
    } catch (MutationsRejectedException e) {
        e.printStackTrace();
    }
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
public static Mutation toMutation(Metric metric) {
    final Mutation mutation = new Mutation(encodeRowKey(metric));

    List<Tag> tags = metric.getTags();
    Collections.sort(tags);

    for (final Tag entry : tags) {
        if (entry.getKey().equals(VISIBILITY_TAG))
            continue;

        final String cf = entry.join();
        // @formatter:off
        String cq = tags.stream().filter(inner -> !inner.equals(entry))
                .filter(inner -> !inner.getKey().equals(VISIBILITY_TAG))
                .map(Tag::join)
                .collect(Collectors.joining(","));
        // @formatter:on

        mutation.put(cf, cq, extractVisibility(tags), metric.getValue().getTimestamp(), extractValue(metric));
    }
    return mutation;
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Test
public void testToMutations() {
    Meta one = new Meta("sys.cpu.user", "tag1", "value1");
    Meta two = new Meta("sys.cpu.user", "tag2", "value2");
    Meta three = new Meta("sys.cpu.user", "tag3", "value3");
    MetaKeySet mks = new MetaKeySet();
    mks.addAll(one.toKeys());
    mks.addAll(two.toKeys());
    mks.addAll(three.toKeys());
    List<Mutation> muts = mks.toMutations();
    Mutation e1 = new Mutation("m:sys.cpu.user");
    e1.put("", "", MetaKeySet.NULL_VALUE);
    Mutation e2 = new Mutation("t:sys.cpu.user");
    e2.put("tag1", "", MetaKeySet.NULL_VALUE);
    e2.put("tag2", "", MetaKeySet.NULL_VALUE);
    e2.put("tag3", "", MetaKeySet.NULL_VALUE);
    Mutation e3 = new Mutation("v:sys.cpu.user");
    e3.put("tag1", "value1", MetaKeySet.NULL_VALUE);
    e3.put("tag2", "value2", MetaKeySet.NULL_VALUE);
    e3.put("tag3", "value3", MetaKeySet.NULL_VALUE);
    Assert.assertEquals(3, muts.size());
    Assert.assertTrue(muts.contains(e1));
    Assert.assertTrue(muts.contains(e2));
    Assert.assertTrue(muts.contains(e3));
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
private int putArray(Mutation m, int count, Object o, Pair<Text, Text> col, String fieldName) {

    // First of all we delete array field on accumulo store
    Text rowKey = new Text(m.getRow());
    Query<K, T> query = newQuery();
    query.setFields(fieldName);
    query.setStartKey((K)rowKey.toString());
    query.setEndKey((K)rowKey.toString());
    deleteByQuery(query);
    flush();
    if (o == null){
      return 0;
    }

    List<?> array = (List<?>) o;  // both GenericArray and DirtyListWrapper
    int j = 0;
    for (Object item : array) {
      m.put(col.getFirst(), new Text(toBytes(j++)), new Value(toBytes(item)));
      count++;
    }
    return count;
  }
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
/**
 * creates output format to write data from flink DataSet to accumulo
 * @return
 * @throws AccumuloSecurityException
 */
public HadoopOutputFormat getHadoopOF() throws AccumuloSecurityException, IOException {

    if(job == null){
        job = Job.getInstance(new Configuration(), jobName);
    }
    AccumuloOutputFormat.setConnectorInfo(job, accumuloUser, new PasswordToken(accumuloPassword));
    ClientConfiguration clientConfig = new ClientConfiguration();
    clientConfig.withInstance(accumuloInstanceName);
    clientConfig.withZkHosts(accumuloZookeeper);
    AccumuloOutputFormat.setZooKeeperInstance(job, clientConfig);
    AccumuloOutputFormat.setDefaultTableName(job, outTable);
    AccumuloFileOutputFormat.setOutputPath(job,new Path("/tmp"));

    HadoopOutputFormat<Text, Mutation> hadoopOF =
            new HadoopOutputFormat<>(new AccumuloOutputFormat() , job);
    return hadoopOF;
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Test
public void testAmc() throws TableExistsException, AccumuloSecurityException, AccumuloException, TableNotFoundException {
    Connector conn = amc.getConnector();
    System.out.println("I am connected as: " + conn.whoami());

    conn.tableOperations().create("TestTable");
    Mutation m1 = new Mutation("row1");
    String testString = "42";
    m1.put("CF", "CQ", testString);
    BatchWriter bw = conn.createBatchWriter("TestTable", new BatchWriterConfig());
    bw.addMutation(m1);
    bw.close();

    Scanner s = conn.createScanner("TestTable", new Authorizations("standard"));
    for(Map.Entry<Key, Value> entry: s){
        System.out.println(entry.getKey());
        System.out.println(entry.getValue());
        assertEquals(entry.getValue().toString(), testString);
    }

    s.close();
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
public void reduce(Iterable<Tuple3<String, Double, Integer>> values, Collector<Tuple2<Text, Mutation>> out) throws Exception {
    for (Tuple3<String,Double,Integer> entry: values) {
        if(i == 0){
            toReturn = "[" + entry.f0.toString();
        }else{
            toReturn = toReturn + "," + entry.f0.toString();
        }
        i++;
        if(i == 10){
            break;
        }

    }
    toReturn += "]";

    Mutation m = new Mutation(rowkey);
    m.put(rowkey, "", new Value(toReturn.getBytes()));
    out.collect(new Tuple2<>(new Text("HighScore"), m));


}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
/**
 * this is called for each token in tweet
 */
public void invoke(Tuple2<TermIndexKey, Integer> value) throws Exception {
    // if the writer isnt already instantiated, do it now
    if (writerForTerms == null) {
        writerForTerms = createBatchWriter(tableTerms);
    }

    //Write termData once per token/user
    Mutation mutationTerms = new Mutation(value._1.getTermBytes());
    if(value._2()<=1)
    {
        //put without value
        // column family, column qualifier without value
        mutationTerms.put(value._1.getSourceBytes(), value._1.rawTwitterDataKey.keyBytes, EMPTY_BYTES);
    }
    else
    {
        //put with number of occurrences
        // column family, column qualifier and value to write
        mutationTerms.put(value._1.getSourceBytes(), value._1.rawTwitterDataKey.keyBytes, (""+value._2()).getBytes());
    }
    writerForTerms.addMutation(mutationTerms);
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
/**
 * this is called for each tweet
 */
public void invoke(Tuple2<RawTwitterDataKey, GeoTemporalKey> value) throws Exception {
    // if the writer isnt already instantiated, do it now
    if (writerForGeoTempIndex == null) {
        writerForGeoTempIndex = createBatchWriter(tableTerms);
    }

    //Write geoTemporalIndex once per tweet
    Mutation mutationTerms = new Mutation(value._2.rowBytes);

    mutationTerms.put(value._1.keyBytes, value._2.columQualifier, EMPTY_BYTES);
    writerForGeoTempIndex.addMutation(mutationTerms);
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
/**
 * this is called for each tweet, writes rawTwitterData into db
 */
public void invoke(Tuple2<RawTwitterDataKey, String> value) throws Exception {
    // if the writer isnt already instantiated, do it now
    if (writerForRawData == null) {
        writerForRawData = createBatchWriter(tableRawData);
    }

    // bytes of the string containing tweet's json
    byte[] tweet = value._2().getBytes();

    //Write raw data once per tweet
    Mutation mutationRawData = new Mutation(value._1.keyBytes);
    // column family, column qualifier and tweet as string
    mutationRawData.put(T_BYTES, EMPTY_BYTES, tweet);

    writerForRawData.addMutation(mutationRawData);
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  Job job = Job.getInstance(getConf());
  job.setJobName(this.getClass().getSimpleName());
  job.setJarByClass(this.getClass());

  Opts opts = new Opts();
  opts.parseArgs(CharacterHistogram.class.getName(), args);

  job.setInputFormatClass(ChunkInputFormat.class);
  opts.setAccumuloConfigs(job);
  job.getConfiguration().set(VIS, opts.visibilities.toString());

  job.setMapperClass(HistMapper.class);
  job.setMapOutputKeyClass(Text.class);
  job.setMapOutputValueClass(Mutation.class);

  job.setNumReduceTasks(0);

  job.setOutputFormatClass(AccumuloOutputFormat.class);

  job.waitForCompletion(true);
  return job.isSuccessful() ? 0 : 1;
}
 

import org.apache.accumulo.core.data.Mutation; //导入依赖的package包/类
public static void main(String[] args) throws AccumuloException, AccumuloSecurityException, MutationsRejectedException, TableExistsException,
    TableNotFoundException {
  ClientOnRequiredTable opts = new ClientOnRequiredTable();
  BatchWriterOpts bwOpts = new BatchWriterOpts();
  opts.parseArgs(InsertWithBatchWriter.class.getName(), args, bwOpts);

  Connector connector = opts.getConnector();
  MultiTableBatchWriter mtbw = connector.createMultiTableBatchWriter(bwOpts.getBatchWriterConfig());

  if (!connector.tableOperations().exists(opts.getTableName()))
    connector.tableOperations().create(opts.getTableName());
  BatchWriter bw = mtbw.getBatchWriter(opts.getTableName());

  Text colf = new Text("colfam");
  System.out.println("writing ...");
  for (int i = 0; i < 10000; i++) {
    Mutation m = new Mutation(new Text(String.format("row_%d", i)));
    for (int j = 0; j < 5; j++) {
      m.put(colf, new Text(String.format("colqual_%d", j)), new Value((String.format("value_%d_%d", i, j)).getBytes()));
    }
    bw.addMutation(m);
    if (i % 100 == 0)
      System.out.println(i);
  }
  mtbw.close();
}