hadoop - 使用 ImportTsv 在 hbase 中插入数据时出现问题

Question

我尝试使用以下命令将数据插入到 hbase 中:

hbase org.apache.hadoop.hbase.mapreduce.ImportTsv -Dimporttsv.columns=HBASE_ROW_KEY,f:pageviews,f:visit -Dimporttsv.separator=\001 -Dimporttsv.bulk.output=output modelvar /000000.gz


hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles modelvar output

其中 modelvar 是假设存储数据的最终 hbase 表。 output是Hfiles存放的HDFS路径。现在的问题是我试图插入的数据是配置单元的输出。因此，默认分隔符是 \001 我无法更改。因此，我将 -Dimporttsv.separator= 的值保留为 \001。但是，我们不能将多个字符作为分隔符。那么，如何将数据插入由hive

编写的 hbase中

Answer 1

IMO，你不能在 hadoop 配置中使用 byte。

但是我们可以注意到 org.apache.hadoop.hbase.mapreduce.ImportTsv.SEPARATOR_CONF_KEY 定义的属性 'importtsv.separator' 在 org.apache.hadoop.hbase.mapreduce.ImportTsv:245 中是 Base64 编码

public static Job createSubmittableJob(Configuration conf, String[] args)
  throws IOException, ClassNotFoundException {

    // Support non-XML supported characters
    // by re-encoding the passed separator as a Base64 string.
    String actualSeparator = conf.get(SEPARATOR_CONF_KEY);
    if (actualSeparator != null) {
      conf.set(SEPARATOR_CONF_KEY,
               Base64.encodeBytes(actualSeparator.getBytes()));
    }
    ...
}

在org.apache.hadoop.hbase.mapreduce.ImportTsv:92中解码

protected void doSetup(Context context) {
    Configuration conf = context.getConfiguration();

    // If a custom separator has been used,
    // decode it back from Base64 encoding.
    separator = conf.get(ImportTsv.SEPARATOR_CONF_KEY);
    if (separator == null) {
      separator = ImportTsv.DEFAULT_SEPARATOR;
    } else {
      separator = new String(Base64.decode(separator));
    }
    ...
}

最后在 org.apache.hadoop.hbase.mapreduce.ImportTsv:97 中检查为单个字节

public TsvParser(String columnsSpecification, String separatorStr) {
      // Configure separator
      byte[] separator = Bytes.toBytes(separatorStr);
      Preconditions.checkArgument(separator.length == 1,
        "TsvParser only supports single-byte separators");
      separatorByte = separator[0];
      ...
}

作为解决方案，我建议您重新声明一个 main 方法，该方法在执行前修改配置的属性。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.mapreduce.ImportTsv;

public class ImportTsvByteSeparator extends ImportTsv
{
    /**
     * Main entry point.
     *
     * @param args  The command line parameters.
     * @throws Exception When running the job fails.
     */
    public static void main(String[] args) throws Exception {

      // We just have to modify the configuration
      Configuration conf = HBaseConfiguration.create();
      int byteSeparator = conf.getInt("importtsv.byte_separator", 001);
      String separator = Character.toString((char) byteSeparator);
      conf.set("importtsv.separator", separator);

      // Now we call ImportTsv main's method
      ImportTsv.main(args);
    }
}

由于属性的可见性，我不认为我们可以覆盖进程内部的某些方法(如 createSubmittableJob())。