| 方法名称和描述 |
|
public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException
设置用于作业的输入。它使用给定的输入规范查询 Metastore，并将匹配的分区序列化到 MapReduce 任务的作业配置中。
|
|
public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException
设置用于作业的输入。它使用给定的输入规范查询 Metastore，并将匹配的分区序列化到 MapReduce 任务的作业配置中。
|
|
public HCatInputFormat setFilter(String filter)throws IOException
在输入表上设置过滤器。
|
|
public HCatInputFormat setProperties(Properties properties) throws IOException
为输入格式设置属性。
|
| 方法名称和描述 |
|
public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException
设置有关为作业写入的输出的信息。它查询元数据服务器以查找要用于表的 StorageHandler。如果分区已经发布，则会引发错误。
|
|
public static void setSchema (Configuration conf, HCatSchema schema) throws IOException
为写出到分区的数据设置架构。如果未调用，则默认情况下将表架构用于分区。
|
|
public RecordWriter <WritableComparable<?>, HCatRecord> getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException
获取作业的记录编写器。它使用 StorageHandler 的默认 OutputFormat 来获取记录编写器。
|
|
public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException
获取此输出格式的输出提交者。它确保正确提交输出。
|
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); } }
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
