001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mapred;
019
020import java.io.IOException;
021import org.apache.hadoop.fs.FileAlreadyExistsException;
022import org.apache.hadoop.fs.FileSystem;
023import org.apache.hadoop.hbase.TableName;
024import org.apache.hadoop.hbase.client.BufferedMutator;
025import org.apache.hadoop.hbase.client.Connection;
026import org.apache.hadoop.hbase.client.ConnectionFactory;
027import org.apache.hadoop.hbase.client.Put;
028import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
029import org.apache.hadoop.mapred.FileOutputFormat;
030import org.apache.hadoop.mapred.InvalidJobConfException;
031import org.apache.hadoop.mapred.JobConf;
032import org.apache.hadoop.mapred.RecordWriter;
033import org.apache.hadoop.mapred.Reporter;
034import org.apache.hadoop.util.Progressable;
035import org.apache.yetus.audience.InterfaceAudience;
036
037/**
038 * Convert Map/Reduce output and write it to an HBase table
039 */
040@InterfaceAudience.Public
041public class TableOutputFormat extends FileOutputFormat<ImmutableBytesWritable, Put> {
042
043  /** JobConf parameter that specifies the output table */
044  public static final String OUTPUT_TABLE = "hbase.mapred.outputtable";
045
046  /**
047   * Convert Reduce output (key, value) to (HStoreKey, KeyedDataArrayWritable) and write to an HBase
048   * table.
049   */
050  protected static class TableRecordWriter implements RecordWriter<ImmutableBytesWritable, Put> {
051    private BufferedMutator m_mutator;
052    private Connection conn;
053
054    /**
055     * Instantiate a TableRecordWriter with the HBase HClient for writing.
056     * @deprecated since 2.0.0 and will be removed in 3.0.0. Please use
057     *             {@code #TableRecordWriter(JobConf)} instead. This version does not clean up
058     *             connections and will leak connections (removed in 2.0).
059     * @see <a href="https://issues.apache.org/jira/browse/HBASE-16774">HBASE-16774</a>
060     */
061    @Deprecated
062    public TableRecordWriter(final BufferedMutator mutator) throws IOException {
063      this.m_mutator = mutator;
064      this.conn = null;
065    }
066
067    /**
068     * Instantiate a TableRecordWriter with a BufferedMutator for batch writing.
069     */
070    public TableRecordWriter(JobConf job) throws IOException {
071      // expecting exactly one path
072      TableName tableName = TableName.valueOf(job.get(OUTPUT_TABLE));
073      try {
074        this.conn = ConnectionFactory.createConnection(job);
075        this.m_mutator = conn.getBufferedMutator(tableName);
076      } finally {
077        if (this.m_mutator == null) {
078          conn.close();
079          conn = null;
080        }
081      }
082    }
083
084    public void close(Reporter reporter) throws IOException {
085      try {
086        if (this.m_mutator != null) {
087          this.m_mutator.close();
088        }
089      } finally {
090        if (conn != null) {
091          this.conn.close();
092        }
093      }
094    }
095
096    public void write(ImmutableBytesWritable key, Put value) throws IOException {
097      m_mutator.mutate(new Put(value));
098    }
099  }
100
101  /**
102   * Creates a new record writer. Be aware that the baseline javadoc gives the impression that there
103   * is a single {@link RecordWriter} per job but in HBase, it is more natural if we give you a new
104   * RecordWriter per call of this method. You must close the returned RecordWriter when done.
105   * Failure to do so will drop writes.
106   * @param ignored Ignored filesystem
107   * @param job     Current JobConf
108   * @param name    Name of the job n * @return The newly created writer instance.
109   * @throws IOException When creating the writer fails.
110   */
111  @Override
112  public RecordWriter getRecordWriter(FileSystem ignored, JobConf job, String name,
113    Progressable progress) throws IOException {
114    // Clear write buffer on fail is true by default so no need to reset it.
115    return new TableRecordWriter(job);
116  }
117
118  @Override
119  public void checkOutputSpecs(FileSystem ignored, JobConf job)
120    throws FileAlreadyExistsException, InvalidJobConfException, IOException {
121    String tableName = job.get(OUTPUT_TABLE);
122    if (tableName == null) {
123      throw new IOException("Must specify table name");
124    }
125  }
126}