HCatalog-入出力フォーマット

ザ・ HCatInputFormat そして HCatOutputFormatインターフェイスはHDFSからデータを読み取るために使用され、処理後、MapReduceジョブを使用して結果のデータをHDFSに書き込みます。入力形式と出力形式のインターフェイスについて詳しく説明します。

HCatInputFormat

ザ・ HCatInputFormatMapReduceジョブとともに使用され、HCatalog管理テーブルからデータを読み取ります。HCatInputFormatは、データがテーブルに公開されているかのようにデータを読み取るためのHadoop 0.20 MapReduceAPIを公開します。

シニア番号	メソッド名と説明
1	public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException ジョブに使用する入力を設定します。指定された入力仕様でメタストアにクエリを実行し、一致するパーティションをMapReduceタスクのジョブ構成にシリアル化します。
2	public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException ジョブに使用する入力を設定します。指定された入力仕様でメタストアにクエリを実行し、一致するパーティションをMapReduceタスクのジョブ構成にシリアル化します。
3	public HCatInputFormat setFilter(String filter)throws IOException 入力テーブルにフィルターを設定します。
4	public HCatInputFormat setProperties(Properties properties) throws IOException 入力フォーマットのプロパティを設定します。

ザ・ HCatInputFormat APIには次のメソッドが含まれています-

setInput
setOutputSchema
getTableSchema

使用するには HCatInputFormat データを読み取るには、最初にインスタンス化します InputJobInfo テーブルから必要な情報を読み取ってから、 setInput とともに InputJobInfo。

あなたは使用することができます setOutputSchema 含める方法 projection schema、出力フィールドを指定します。スキーマが指定されていない場合、テーブル内のすべての列が返されます。getTableSchemaメソッドを使用して、指定した入力テーブルのテーブルスキーマを決定できます。

HCatOutputFormat

HCatOutputFormatは、MapReduceジョブとともに使用され、HCatalog管理テーブルにデータを書き込みます。HCatOutputFormatは、データをテーブルに書き込むためのHadoop 0.20 MapReduceAPIを公開します。MapReduceジョブがHCatOutputFormatを使用して出力を書き込む場合、テーブルに構成されているデフォルトのOutputFormatが使用され、ジョブの完了後に新しいパーティションがテーブルに公開されます。

シニア番号	メソッド名と説明
1	public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException ジョブに書き込む出力に関する情報を設定します。メタデータサーバーにクエリを実行して、テーブルに使用するStorageHandlerを見つけます。パーティションがすでに公開されている場合は、エラーがスローされます。
2	public static void setSchema (Configuration conf, HCatSchema schema) throws IOException パーティションに書き出されるデータのスキーマを設定します。これが呼び出されない場合、パーティションにはデフォルトでテーブルスキーマが使用されます。
3	public RecordWriter <WritableComparable<?>, HCatRecord > getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException その仕事のレコードライターを入手してください。StorageHandlerのデフォルトのOutputFormatを使用して、レコードライターを取得します。
4	public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException この出力形式の出力コミッターを取得します。これにより、出力が正しくコミットされます。

ザ・ HCatOutputFormat APIには次のメソッドが含まれています-

setOutput
setSchema
getTableSchema

HCatOutputFormatの最初の呼び出しは、 setOutput; 他の呼び出しは、出力フォーマットが初期化されていないことを示す例外をスローします。

書き出されるデータのスキーマは、 setSchema方法。書き込むデータのスキーマを指定して、このメソッドを呼び出す必要があります。データのスキーマがテーブルスキーマと同じである場合は、次を使用できます。HCatOutputFormat.getTableSchema() テーブルスキーマを取得し、それをに渡します setSchema()。

例

次のMapReduceプログラムは、2番目の列（「列1」）に整数があると想定する1つのテーブルからデータを読み取り、検出した各個別の値のインスタンスの数をカウントします。つまり、「select col1, count(*) from $table group by col1;"。

たとえば、2番目の列の値が{1、1、1、3、3、5}の場合、プログラムは次の値とカウントの出力を生成します-

1, 3
3, 2
5, 1

プログラムコードを見てみましょう-

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import org.apache.HCatalog.common.HCatConstants;
import org.apache.HCatalog.data.DefaultHCatRecord;
import org.apache.HCatalog.data.HCatRecord;
import org.apache.HCatalog.data.schema.HCatSchema;

import org.apache.HCatalog.mapreduce.HCatInputFormat;
import org.apache.HCatalog.mapreduce.HCatOutputFormat;
import org.apache.HCatalog.mapreduce.InputJobInfo;
import org.apache.HCatalog.mapreduce.OutputJobInfo;

public class GroupByAge extends Configured implements Tool {

   public static class Map extends Mapper<WritableComparable, 
      HCatRecord, IntWritable, IntWritable> {
      int age;
		
      @Override
      protected void map(
         WritableComparable key, HCatRecord value,
         org.apache.hadoop.mapreduce.Mapper<WritableComparable,
         HCatRecord, IntWritable, IntWritable>.Context context
      )throws IOException, InterruptedException {
         age = (Integer) value.get(1);
         context.write(new IntWritable(age), new IntWritable(1));
      }
   }
	
   public static class Reduce extends Reducer<IntWritable, IntWritable,
      WritableComparable, HCatRecord> {
      @Override
      protected void reduce(
         IntWritable key, java.lang.Iterable<IntWritable> values,
         org.apache.hadoop.mapreduce.Reducer<IntWritable, IntWritable,
         WritableComparable, HCatRecord>.Context context
      )throws IOException ,InterruptedException {
         int sum = 0;
         Iterator<IntWritable> iter = values.iterator();
			
         while (iter.hasNext()) {
            sum++;
            iter.next();
         }
			
         HCatRecord record = new DefaultHCatRecord(2);
         record.set(0, key.get());
         record.set(1, sum);
         context.write(null, record);
      }
   }
	
   public int run(String[] args) throws Exception {
      Configuration conf = getConf();
      args = new GenericOptionsParser(conf, args).getRemainingArgs();
		
      String serverUri = args[0];
      String inputTableName = args[1];
      String outputTableName = args[2];
      String dbName = null;
      String principalID = System
		
      .getProperty(HCatConstants.HCAT_METASTORE_PRINCIPAL);
      if (principalID != null)
      conf.set(HCatConstants.HCAT_METASTORE_PRINCIPAL, principalID);
      Job job = new Job(conf, "GroupByAge");
      HCatInputFormat.setInput(job, InputJobInfo.create(dbName, inputTableName, null));

      // initialize HCatOutputFormat
      job.setInputFormatClass(HCatInputFormat.class);
      job.setJarByClass(GroupByAge.class);
      job.setMapperClass(Map.class);
      job.setReducerClass(Reduce.class);
		
      job.setMapOutputKeyClass(IntWritable.class);
      job.setMapOutputValueClass(IntWritable.class);
      job.setOutputKeyClass(WritableComparable.class);
      job.setOutputValueClass(DefaultHCatRecord.class);
		
      HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName, outputTableName, null));
      HCatSchema s = HCatOutputFormat.getTableSchema(job);
      System.err.println("INFO: output schema explicitly set for writing:" + s);
      HCatOutputFormat.setSchema(job, s);
      job.setOutputFormatClass(HCatOutputFormat.class);
      return (job.waitForCompletion(true) ? 0 : 1);
   }
	
   public static void main(String[] args) throws Exception {
      int exitCode = ToolRunner.run(new GroupByAge(), args);
      System.exit(exitCode);
   }
}

上記のプログラムをコンパイルする前に、いくつかをダウンロードする必要があります jars それらをに追加します classpathこのアプリケーションの場合。すべてのHivejarとHCatalogjarをダウンロードする必要があります（HCatalog-core-0.5.0.jar、hive-metastore-0.10.0.jar、libthrift-0.7.0.jar、hive-exec-0.10.0.jar、 libfb303-0.7.0.jar、jdo2-api-2.3-ec.jar、slf4j-api-1.6.1.jar）。

次のコマンドを使用してそれらをコピーします jar からのファイル local に HDFS それらをに追加します classpath。

bin/hadoop fs -copyFromLocal $HCAT_HOME/share/HCatalog/HCatalog-core-0.5.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/hive-metastore-0.10.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/libthrift-0.7.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/hive-exec-0.10.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/libfb303-0.7.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/jdo2-api-2.3-ec.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/slf4j-api-1.6.1.jar /tmp

export LIB_JARS=hdfs:///tmp/HCatalog-core-0.5.0.jar,
hdfs:///tmp/hive-metastore-0.10.0.jar,
hdfs:///tmp/libthrift-0.7.0.jar,
hdfs:///tmp/hive-exec-0.10.0.jar,
hdfs:///tmp/libfb303-0.7.0.jar,
hdfs:///tmp/jdo2-api-2.3-ec.jar,
hdfs:///tmp/slf4j-api-1.6.1.jar

次のコマンドを使用して、指定されたプログラムをコンパイルして実行します。

$HADOOP_HOME/bin/hadoop jar GroupByAge tmp/hive

次に、出力ディレクトリ（hdfs：user / tmp / hive）で出力（part_0000、part_0001）を確認します。