HCatalog - Eingabe Ausgabeformat
Das HCatInputFormat und HCatOutputFormatSchnittstellen werden zum Lesen von Daten aus HDFS verwendet und schreiben die resultierenden Daten nach der Verarbeitung mithilfe des MapReduce-Jobs in HDFS. Lassen Sie uns die Schnittstellen für das Eingabe- und Ausgabeformat näher erläutern.
HCatInputFormat
Das HCatInputFormatwird mit MapReduce-Jobs verwendet, um Daten aus von HCatalog verwalteten Tabellen zu lesen. HCatInputFormat stellt eine Hadoop 0.20 MapReduce-API zum Lesen von Daten bereit, als ob sie in einer Tabelle veröffentlicht worden wären.
| Sr.Nr. | Methodenname & Beschreibung |
|---|---|
| 1 | public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException Stellen Sie die für den Job zu verwendenden Eingaben ein. Es fragt den Metastore mit der angegebenen Eingabespezifikation ab und serialisiert übereinstimmende Partitionen in die Jobkonfiguration für MapReduce-Aufgaben. |
| 2 | public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException Stellen Sie die für den Job zu verwendenden Eingaben ein. Es fragt den Metastore mit der angegebenen Eingabespezifikation ab und serialisiert übereinstimmende Partitionen in die Jobkonfiguration für MapReduce-Aufgaben. |
| 3 | public HCatInputFormat setFilter(String filter)throws IOException Setzen Sie einen Filter in der Eingabetabelle. |
| 4 | public HCatInputFormat setProperties(Properties properties) throws IOException Legen Sie die Eigenschaften für das Eingabeformat fest. |
Das HCatInputFormat Die API enthält die folgenden Methoden:
- setInput
- setOutputSchema
- getTableSchema
Benutzen HCatInputFormat Um Daten zu lesen, instanziieren Sie zuerst eine InputJobInfo mit den notwendigen Informationen aus der Tabelle gelesen und dann aufgerufen setInput mit dem InputJobInfo.
Du kannst den ... benutzen setOutputSchema Methode, um a einzuschließen projection schema, um die Ausgabefelder anzugeben. Wenn kein Schema angegeben wird, werden alle Spalten in der Tabelle zurückgegeben. Mit der Methode getTableSchema können Sie das Tabellenschema für eine angegebene Eingabetabelle ermitteln.
HCatOutputFormat
HCatOutputFormat wird mit MapReduce-Jobs verwendet, um Daten in von HCatalog verwaltete Tabellen zu schreiben. HCatOutputFormat macht eine Hadoop 0.20 MapReduce-API zum Schreiben von Daten in eine Tabelle verfügbar. Wenn ein MapReduce-Job HCatOutputFormat zum Schreiben der Ausgabe verwendet, wird das für die Tabelle konfigurierte Standard-OutputFormat verwendet und die neue Partition wird nach Abschluss des Jobs in der Tabelle veröffentlicht.
| Sr.Nr. | Methodenname & Beschreibung |
|---|---|
| 1 | public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException Legen Sie die Informationen zu der Ausgabe fest, die für den Job geschrieben werden sollen. Es fragt den Metadatenserver ab, um den StorageHandler zu finden, der für die Tabelle verwendet werden soll. Es wird ein Fehler ausgegeben, wenn die Partition bereits veröffentlicht ist. |
| 2 | public static void setSchema (Configuration conf, HCatSchema schema) throws IOException Legen Sie das Schema für die Daten fest, die auf die Partition geschrieben werden. Das Tabellenschema wird standardmäßig für die Partition verwendet, wenn dies nicht aufgerufen wird. |
| 3 | public RecordWriter <WritableComparable<?>, HCatRecord > getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException Holen Sie sich den Plattenschreiber für den Job. Es verwendet das Standard-OutputFormat des StorageHandlers, um den Datensatzschreiber abzurufen. |
| 4 | public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException Holen Sie sich den Ausgabe-Committer für dieses Ausgabeformat. Es stellt sicher, dass die Ausgabe korrekt festgeschrieben wird. |
Das HCatOutputFormat Die API enthält die folgenden Methoden:
- setOutput
- setSchema
- getTableSchema
Der erste Aufruf des HCatOutputFormat muss sein setOutput;; Bei jedem anderen Aufruf wird eine Ausnahme ausgelöst, die besagt, dass das Ausgabeformat nicht initialisiert ist.
Das Schema für die auszuschreibenden Daten wird durch das angegeben setSchemaMethode. Sie müssen diese Methode aufrufen und das Schema der Daten angeben, die Sie schreiben. Wenn Ihre Daten dasselbe Schema wie das Tabellenschema haben, können Sie verwendenHCatOutputFormat.getTableSchema() um das Tabellenschema zu erhalten und es dann an weiterzugeben setSchema().
Beispiel
Das folgende MapReduce-Programm liest Daten aus einer Tabelle, von der angenommen wird, dass sie eine Ganzzahl in der zweiten Spalte ("Spalte 1") enthält, und zählt, wie viele Instanzen jedes einzelnen Werts gefunden werden. Das heißt, es entspricht "select col1, count(*) from $table group by col1;".
Wenn die Werte in der zweiten Spalte beispielsweise {1, 1, 1, 3, 3, 5} sind, erzeugt das Programm die folgende Ausgabe von Werten und Zählwerten:
1, 3
3, 2
5, 1Schauen wir uns jetzt den Programmcode an -
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);
}
}Bevor Sie das obige Programm kompilieren können, müssen Sie einige herunterladen jars und fügen Sie diese dem hinzu classpathfür diese Anwendung. Sie müssen alle Hive- und HCatalog-Gläser herunterladen (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).
Verwenden Sie die folgenden Befehle, um diese zu kopieren jar Dateien aus local zu HDFS und fügen Sie diese dem hinzu 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.jarVerwenden Sie den folgenden Befehl, um das angegebene Programm zu kompilieren und auszuführen.
$HADOOP_HOME/bin/hadoop jar GroupByAge tmp/hiveÜberprüfen Sie nun Ihr Ausgabeverzeichnis (hdfs: user / tmp / hive) auf die Ausgabe (part_0000, part_0001).