import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p>
 * Handles encryption of the search values as needed, filtering for the given column on the server or client side as appropriate.
 *
 * @throws EncryptionException
 *           The reason for the failure can be retrieved by calling {@link EncryptionException#getCause()}.
 */
@Override
public void fetchColumn(Text colFam, Text colQual) {
  checkArgument(colFam != null, "colFam is null");
  checkArgument(colQual != null, "colQual is null");

  ColumnFilterResult search = encryptor.getColumnFilter(colFam, colQual);
  for (Column column : search.serverSideFilters) {
    if (column.getColumnQualifier() == null) {
      scanner.fetchColumnFamily(new Text(column.getColumnFamily()));
    } else {
      scanner.fetchColumn(new Text(column.getColumnFamily()), new Text(column.getColumnQualifier()));
    }
  }
  if (search.needsClientSideFiltering) {
    clientSideColumnFilters.add(new Column(TextUtil.getBytes(colFam), TextUtil.getBytes(colQual), null));
  }
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
private void setupSplitsFile(final Job job, final TableOperations parentTableOperations, final String parentTableName, final String childTableName) throws Exception {
    final FileSystem fs = FileSystem.get(conf);
    fs.setPermission(getPath(baseOutputDir, childTableName), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
    final Path splitsPath = getPath(baseOutputDir, childTableName, "splits.txt");
    final Collection<Text> splits = parentTableOperations.listSplits(parentTableName, 100);
    log.info("Creating splits file at: " + splitsPath);
    try (PrintStream out = new PrintStream(new BufferedOutputStream(fs.create(splitsPath)), false, StandardCharsets.UTF_8.name())) {
        for (final Text split : splits) {
            final String encoded = new String(Base64.encodeBase64(TextUtil.getBytes(split)), StandardCharsets.UTF_8);
            out.println(encoded);
        }
    }
    fs.setPermission(splitsPath, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));

    final String userDir = PathUtils.clean(System.getProperty("user.dir"));
    // The splits file has a symlink created in the user directory for some reason.
    // It might be better to copy the entire file for Windows but it doesn't seem to matter if
    // the user directory symlink is broken.
    java.nio.file.Files.deleteIfExists(new File(userDir, "splits.txt").toPath());
    //Files.copy(new File(splitsPath.toString()), new File(userDir, "splits.txt"));
    job.setPartitionerClass(KeyRangePartitioner.class);
    KeyRangePartitioner.setSplitFile(job, splitsPath.toString());
    job.setNumReduceTasks(splits.size() + 1);
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p>
 * Handles encryption of the search values as needed, filtering for the given column family on the server or client side as appropriate.
 *
 * @throws EncryptionException
 *           The reason for the failure can be retrieved by calling {@link EncryptionException#getCause()}.
 */
@Override
public void fetchColumnFamily(Text col) {
  checkArgument(col != null, "col is null");

  ColumnFilterResult search = encryptor.getColumnFamilyFilter(col);
  for (Column filter : search.serverSideFilters) {
    scanner.fetchColumnFamily(new Text(filter.getColumnFamily()));
  }
  if (search.needsClientSideFiltering) {
    clientSideColumnFilters.add(new Column(TextUtil.getBytes(col), null, null));
  }
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * Get the filter for the given value.
 *
 * @param col
 *          Column family to filter.
 * @return Results with columns to filter server ide and whether client side filtering is needed.
 */
public ColumnFilterResult getColumnFamilyFilter(Text col) {
  checkArgument(col != null, "col is null");

  MutableEntry key = new MutableEntry();
  key.colF = TextUtil.getBytes(col);

  return getColumnFamilyFilter(key, ImmutableSet.of(EntryField.COLUMN_FAMILY));
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * Encode a <code>Text</code> array of all the columns to intersect on
 * 
 * @param columns
 *          The columns to be encoded
 * @return A Base64 encoded string (using a \n delimiter) of all columns to intersect on.
 */
public static String encodeColumns(Text[] columns) {
  StringBuilder sb = new StringBuilder();
  for (int i = 0; i < columns.length; i++) {
    sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(columns[i]))));
    sb.append('\n');
  }
  return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * Encode a <code>Text</code> array of all of the terms to intersect on. The terms should match the columns in a one-to-one manner
 * 
 * @param terms
 *          The terms to be encoded
 * @return A Base64 encoded string (using a \n delimiter) of all terms to intersect on.
 */
public static String encodeTermValues(Text[] terms) {
  StringBuilder sb = new StringBuilder();
  for (int i = 0; i < terms.length; i++) {
    sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(terms[i]))));
    sb.append('\n');
  }
  
  return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
protected static String encodeColumns(TextColumn[] columns) {
  StringBuilder sb = new StringBuilder();
  for (int i = 0; i < columns.length; i++) {
    sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(columns[i].getColumnFamily())), StandardCharsets.UTF_8));
    sb.append('\n');
    sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(columns[i].getColumnQualifier())), StandardCharsets.UTF_8));
    sb.append('\u0001');
  }
  return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * @param columns
 * @return encoded columns
 * @deprecated since 1.4. To be made protected. Do not interact with flags string directly, just use
 *             {@link #setColumnFamilies(IteratorSetting, Text[], boolean[])}.
 */
@Deprecated
   public static String encodeColumns(final Text[] columns) {
	final StringBuilder sb = new StringBuilder();
	for (final Text column : columns) {
		sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(column)), StandardCharsets.UTF_8));
		sb.append('\n');
	}
	return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * Encode a <code>Text</code> array of all the columns to intersect on
 *
 * @param columns The columns to be encoded
 * @return A Base64 encoded string (using a \n delimiter) of all columns to intersect on.
 */
public static String encodeColumns(Text[] columns) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < columns.length; i++) {
        sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(columns[i]))));
        sb.append('\n');
    }
    return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
/**
 * Encode a <code>Text</code> array of all of the terms to intersect on. The terms should match the columns in a one-to-one manner
 *
 * @param terms The terms to be encoded
 * @return A Base64 encoded string (using a \n delimiter) of all terms to intersect on.
 */
public static String encodeTermValues(Text[] terms) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < terms.length; i++) {
        sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(terms[i]))));
        sb.append('\n');
    }

    return sb.toString();
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
@Override
public int run(String[] args) {
  Opts opts = new Opts();
  opts.parseArgs(BulkIngestExample.class.getName(), args);

  Configuration conf = getConf();
  PrintStream out = null;
  try {
    Job job = Job.getInstance(conf);
    job.setJobName("bulk ingest example");
    job.setJarByClass(this.getClass());

    job.setInputFormatClass(TextInputFormat.class);

    job.setMapperClass(MapClass.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);

    job.setReducerClass(ReduceClass.class);
    job.setOutputFormatClass(AccumuloFileOutputFormat.class);
    opts.setAccumuloConfigs(job);

    Connector connector = opts.getConnector();

    TextInputFormat.setInputPaths(job, new Path(opts.inputDir));
    AccumuloFileOutputFormat.setOutputPath(job, new Path(opts.workDir + "/files"));

    FileSystem fs = FileSystem.get(conf);
    out = new PrintStream(new BufferedOutputStream(fs.create(new Path(opts.workDir + "/splits.txt"))));

    Collection<Text> splits = connector.tableOperations().listSplits(opts.getTableName(), 100);
    for (Text split : splits)
      out.println(Base64.getEncoder().encodeToString(TextUtil.getBytes(split)));

    job.setNumReduceTasks(splits.size() + 1);
    out.close();

    job.setPartitionerClass(RangePartitioner.class);
    RangePartitioner.setSplitFile(job, opts.workDir + "/splits.txt");

    job.waitForCompletion(true);
    Path failures = new Path(opts.workDir, "failures");
    fs.delete(failures, true);
    fs.mkdirs(new Path(opts.workDir, "failures"));
    // With HDFS permissions on, we need to make sure the Accumulo user can read/move the rfiles
    FsShell fsShell = new FsShell(conf);
    fsShell.run(new String[] {"-chmod", "-R", "777", opts.workDir});
    connector.tableOperations().importDirectory(opts.getTableName(), opts.workDir + "/files", opts.workDir + "/failures", false);

  } catch (Exception e) {
    throw new RuntimeException(e);
  } finally {
    if (out != null)
      out.close();
  }

  return 0;
}
 

import org.apache.accumulo.core.util.TextUtil; //导入依赖的package包/类
@Override
public byte[] encode(Text data) {
  return TextUtil.getBytes(data);
}