HCatalog - Formato de entrada y salida
los HCatInputFormat y HCatOutputFormatLas interfaces se utilizan para leer datos de HDFS y, después del procesamiento, escribir los datos resultantes en HDFS utilizando el trabajo MapReduce. Elaboremos las interfaces de formato de entrada y salida.
HCatInputFormat
los HCatInputFormatse usa con trabajos de MapReduce para leer datos de tablas administradas por HCatalog. HCatInputFormat expone una API MapReduce de Hadoop 0.20 para leer datos como si se hubieran publicado en una tabla.
| No Señor. | Nombre y descripción del método |
|---|---|
| 1 | public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException Configure las entradas que se utilizarán para el trabajo. Consulta la tienda de metadatos con la especificación de entrada dada y serializa las particiones coincidentes en la configuración del trabajo para las tareas de MapReduce. |
| 2 | public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException Configure las entradas que se utilizarán para el trabajo. Consulta la tienda de metadatos con la especificación de entrada dada y serializa las particiones coincidentes en la configuración del trabajo para las tareas de MapReduce. |
| 3 | public HCatInputFormat setFilter(String filter)throws IOException Establezca un filtro en la tabla de entrada. |
| 4 | public HCatInputFormat setProperties(Properties properties) throws IOException Establecer propiedades para el formato de entrada. |
los HCatInputFormat API incluye los siguientes métodos:
- setInput
- setOutputSchema
- getTableSchema
Usar HCatInputFormat para leer datos, primero cree una instancia InputJobInfo con la información necesaria de la tabla que se está leyendo y luego llame setInput con el InputJobInfo.
Puedes usar el setOutputSchema método para incluir un projection schema, para especificar los campos de salida. Si no se especifica un esquema, se devolverán todas las columnas de la tabla. Puede utilizar el método getTableSchema para determinar el esquema de tabla para una tabla de entrada especificada.
HCatOutputFormat
HCatOutputFormat se usa con trabajos de MapReduce para escribir datos en tablas administradas por HCatalog. HCatOutputFormat expone una API MapReduce de Hadoop 0.20 para escribir datos en una tabla. Cuando un trabajo de MapReduce usa HCatOutputFormat para escribir la salida, se usa el OutputFormat predeterminado configurado para la tabla y la nueva partición se publica en la tabla una vez que se completa el trabajo.
| No Señor. | Nombre y descripción del método |
|---|---|
| 1 | public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException Configure la información sobre la salida para escribir para el trabajo. Consulta al servidor de metadatos para encontrar el StorageHandler que se utilizará para la tabla. Lanza un error si la partición ya está publicada. |
| 2 | public static void setSchema (Configuration conf, HCatSchema schema) throws IOException Establezca el esquema para los datos que se escriben en la partición. El esquema de tabla se usa de forma predeterminada para la partición si no se llama. |
| 3 | public RecordWriter <WritableComparable<?>, HCatRecord > getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException Consiga al escritor de discos para el trabajo. Utiliza el OutputFormat predeterminado de StorageHandler para obtener el escritor de registros. |
| 4 | public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException Obtenga el confirmador de salida para este formato de salida. Asegura que la salida se confirme correctamente. |
los HCatOutputFormat API incluye los siguientes métodos:
- setOutput
- setSchema
- getTableSchema
La primera llamada en HCatOutputFormat debe ser setOutput; cualquier otra llamada lanzará una excepción diciendo que el formato de salida no está inicializado.
El esquema de los datos que se escriben lo especifica el setSchemamétodo. Debe llamar a este método, proporcionando el esquema de datos que está escribiendo. Si sus datos tienen el mismo esquema que el esquema de la tabla, puede usarHCatOutputFormat.getTableSchema() para obtener el esquema de la tabla y luego pasarlo a setSchema().
Ejemplo
El siguiente programa MapReduce lee datos de una tabla que asume que tiene un número entero en la segunda columna ("columna 1"), y cuenta cuántas instancias de cada valor distinto encuentra. Es decir, hace el equivalente de "select col1, count(*) from $table group by col1;".
Por ejemplo, si los valores de la segunda columna son {1, 1, 1, 3, 3, 5}, el programa producirá la siguiente salida de valores y recuentos:
1, 3
3, 2
5, 1Echemos ahora un vistazo al código del programa:
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);
}
}Antes de compilar el programa anterior, debe descargar algunos jars y agregarlos al classpathpara esta aplicación. Necesita descargar todos los archivos de Hive y HCatalog (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).
Utilice los siguientes comandos para copiar esos jar archivos de local a HDFS y agregarlos al 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.jarUtilice el siguiente comando para compilar y ejecutar el programa dado.
$HADOOP_HOME/bin/hadoop jar GroupByAge tmp/hiveAhora, verifique su directorio de salida (hdfs: user / tmp / hive) para ver la salida (part_0000, part_0001).