@Override
  public RecordWriter getHiveRecordWriter(
    final JobConf jc,
    final Path finalOutPath,
    Class<? extends Writable> valueClass,
    boolean isCompressed,
    Properties tableProperties,
    final Progressable progressable) throws IOException {

    String hbaseTableName = jc.get(HBaseSerDe.HBASE_TABLE_NAME);
    if (hbaseTableName == null) {
      hbaseTableName = tableProperties.getProperty(hive_metastoreConstants.META_TABLE_NAME);
      hbaseTableName = hbaseTableName.toLowerCase();
      if (hbaseTableName.startsWith(HBaseStorageHandler.DEFAULT_PREFIX)) {
        hbaseTableName = hbaseTableName.substring(HBaseStorageHandler.DEFAULT_PREFIX.length());
      }
    }
    jc.set(OUTPUT_TABLE_NAME_CONF_KEY, hbaseTableName);

    // Read configuration for the target path, first from jobconf, then from table properties
    String hfilePath = getFamilyPath(jc, tableProperties);
    if (hfilePath == null) {
      throw new RuntimeException(
        "Please set " + HFILE_FAMILY_PATH + " to target location for HFiles");
    }

    // Target path's last component is also the column family name.
    final Path columnFamilyPath = new Path(hfilePath);
    final String columnFamilyName = columnFamilyPath.getName();
    final byte [] columnFamilyNameBytes = Bytes.toBytes(columnFamilyName);
    final Job job = new Job(jc);
    setCompressOutput(job, isCompressed);
    setOutputPath(job, finalOutPath);

    // Create the HFile writer
    final org.apache.hadoop.mapreduce.TaskAttemptContext tac =
      ShimLoader.getHadoopShims().newTaskAttemptContext(
          job.getConfiguration(), progressable);

    final Path outputdir = FileOutputFormat.getOutputPath(tac);
    final Path taskAttemptOutputdir = new FileOutputCommitter(outputdir, tac).getWorkPath();
    final org.apache.hadoop.mapreduce.RecordWriter<
      ImmutableBytesWritable, Cell> fileWriter = getFileWriter(tac);

    // Individual columns are going to be pivoted to HBase cells,
    // and for each row, they need to be written out in order
    // of column name, so sort the column names now, creating a
    // mapping to their column position.  However, the first
    // column is interpreted as the row key.
    String columnList = tableProperties.getProperty("columns");
    String [] columnArray = columnList.split(",");
    final SortedMap<byte [], Integer> columnMap =
      new TreeMap<byte [], Integer>(Bytes.BYTES_COMPARATOR);
    int i = 0;
    for (String columnName : columnArray) {
      if (i != 0) {
        columnMap.put(Bytes.toBytes(columnName), i);
      }
      ++i;
    }

    return new RecordWriter() {

      @Override
      public void close(boolean abort) throws IOException {
        try {
          fileWriter.close(null);
          if (abort) {
            return;
          }
          // Move the hfiles file(s) from the task output directory to the
          // location specified by the user.
          FileSystem fs = outputdir.getFileSystem(jc);
          fs.mkdirs(columnFamilyPath);
          Path srcDir = taskAttemptOutputdir;
          FileStatus[] files = null;
          for (; ; ) {
            try {
              files = fs.listStatus(srcDir, FileUtils.STAGING_DIR_PATH_FILTER);
            } catch (FileNotFoundException fnf) {
              LOG.debug("File doesn't exist {} ",srcDir, fnf);
              break;
            }
            if ((files == null) || (files.length == 0)) {
              throw new IOException("No family directories found in " + srcDir);
            }
            if (files.length != 1) {
              throw new IOException("Multiple family directories found in " + srcDir);
            }
            srcDir = files[0].getPath();
            if (srcDir.getName().equals(columnFamilyName)) {
              break;
            }
            if (files[0].isFile()) {
              throw new IOException("No family directories found in " + taskAttemptOutputdir + ". "
                  + "The last component in hfile path should match column family name "
                  + columnFamilyName);
            }
          }
          if (files != null) {
            for (FileStatus regionFile : fs.listStatus(srcDir, FileUtils.STAGING_DIR_PATH_FILTER)) {
              fs.rename(regionFile.getPath(), new Path(columnFamilyPath, regionFile.getPath().getName()));
            }
          }
        } catch (InterruptedException ex) {
          throw new IOException(ex);
        } catch (FileNotFoundException fnf) {
          // Ignore....
          LOG.debug("File doesn't exist.", fnf);
        }
      }

      private void writeText(Text text) throws IOException {
        // Decompose the incoming text row into fields.
        String s = text.toString();
        String [] fields = s.split("\u0001");
        assert(fields.length <= (columnMap.size() + 1));
        // First field is the row key.
        byte [] rowKeyBytes = Bytes.toBytes(fields[0]);
        // Remaining fields are cells addressed by column name within row.
        for (Map.Entry<byte [], Integer> entry : columnMap.entrySet()) {
          byte [] columnNameBytes = entry.getKey();
          int iColumn = entry.getValue();
          String val;
          if (iColumn >= fields.length) {
            // trailing blank field
            val = "";
          } else {
            val = fields[iColumn];
            if ("\\N".equals(val)) {
              // omit nulls
              continue;
            }
          }
          byte [] valBytes = Bytes.toBytes(val);
          KeyValue kv = new KeyValue(
            rowKeyBytes,
            columnFamilyNameBytes,
            columnNameBytes,
            valBytes);
          try {
            fileWriter.write(null, kv);
          } catch (IOException e) {
            LOG.error("Failed while writing row: " + s);
            throw e;
          } catch (InterruptedException ex) {
            throw new IOException(ex);
          }
        }
      }

      private void writePut(PutWritable put) throws IOException {
        ImmutableBytesWritable row = new ImmutableBytesWritable(put.getPut().getRow());
        SortedMap<byte[], List<Cell>> cells = put.getPut().getFamilyCellMap();
        for (Map.Entry<byte[], List<Cell>> entry : cells.entrySet()) {
          Collections.sort(entry.getValue(), new CellComparatorImpl());
          for (Cell c : entry.getValue()) {
            try {
              fileWriter.write(row, KeyValueUtil.copyToNewKeyValue(c));
            } catch (InterruptedException e) {
              throw (InterruptedIOException) new InterruptedIOException().initCause(e);
            }
          }
        }
      }

      @Override
      public void write(Writable w) throws IOException {
        if (w instanceof Text) {
          writeText((Text) w);
        } else if (w instanceof PutWritable) {
          writePut((PutWritable) w);
        } else {
          throw new IOException("Unexpected writable " + w);
        }
      }
    };
  }
Retrieve the family path, first check the JobConf, then the table properties.
@return the family path or null if not specified.
hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHFileOutputFormat.java