HCatalog - Format d'entrée / sortie

le HCatInputFormat et HCatOutputFormatLes interfaces sont utilisées pour lire les données à partir de HDFS et après le traitement, écrire les données résultantes dans HDFS à l'aide du travail MapReduce. Développons les interfaces de format d'entrée et de sortie.

HCatInputFormat

le HCatInputFormatest utilisé avec les travaux MapReduce pour lire les données des tables gérées par HCatalog. HCatInputFormat expose une API MapReduce Hadoop 0.20 pour lire des données comme si elles avaient été publiées dans une table.

N ° Sr.	Nom et description de la méthode
1	public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException Définissez les entrées à utiliser pour le travail. Il interroge le métastore avec la spécification d'entrée donnée et sérialise les partitions correspondantes dans la configuration du travail pour les tâches MapReduce.
2	public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException Définissez les entrées à utiliser pour le travail. Il interroge le métastore avec la spécification d'entrée donnée et sérialise les partitions correspondantes dans la configuration du travail pour les tâches MapReduce.
3	public HCatInputFormat setFilter(String filter)throws IOException Définissez un filtre sur la table d'entrée.
4	public HCatInputFormat setProperties(Properties properties) throws IOException Définissez les propriétés du format d'entrée.

le HCatInputFormat L'API comprend les méthodes suivantes -

setInput
setOutputSchema
getTableSchema

Utiliser HCatInputFormat pour lire des données, instanciez d'abord un InputJobInfo avec les informations nécessaires du tableau en cours de lecture, puis appel setInput avec le InputJobInfo.

Vous pouvez utiliser le setOutputSchema méthode pour inclure un projection schema, pour spécifier les champs de sortie. Si aucun schéma n'est spécifié, toutes les colonnes de la table seront renvoyées. Vous pouvez utiliser la méthode getTableSchema pour déterminer le schéma de table pour une table d'entrée spécifiée.

HCatOutputFormat

HCatOutputFormat est utilisé avec les travaux MapReduce pour écrire des données dans des tables gérées par HCatalog. HCatOutputFormat expose une API MapReduce Hadoop 0.20 pour écrire des données dans une table. Lorsqu'un travail MapReduce utilise HCatOutputFormat pour écrire la sortie, le OutputFormat par défaut configuré pour la table est utilisé et la nouvelle partition est publiée dans la table une fois le travail terminé.

N ° Sr.	Nom et description de la méthode
1	public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException Définissez les informations sur la sortie à écrire pour le travail. Il interroge le serveur de métadonnées pour trouver le StorageHandler à utiliser pour la table. Il génère une erreur si la partition est déjà publiée.
2	public static void setSchema (Configuration conf, HCatSchema schema) throws IOException Définissez le schéma des données écrites sur la partition. Le schéma de table est utilisé par défaut pour la partition s'il n'est pas appelé.
3	public RecordWriter <WritableComparable<?>, HCatRecord > getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException Obtenez le rédacteur de disques pour le travail. Il utilise le OutputFormat par défaut de StorageHandler pour obtenir l'enregistreur d'enregistrement.
4	public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException Obtenez le committer de sortie pour ce format de sortie. Il garantit que la sortie est validée correctement.

le HCatOutputFormat L'API comprend les méthodes suivantes -

setOutput
setSchema
getTableSchema

Le premier appel sur HCatOutputFormat doit être setOutput; tout autre appel lèvera une exception indiquant que le format de sortie n'est pas initialisé.

Le schéma des données en cours d'écriture est spécifié par le setSchemaméthode. Vous devez appeler cette méthode en fournissant le schéma des données que vous écrivez. Si vos données ont le même schéma que le schéma de table, vous pouvez utiliserHCatOutputFormat.getTableSchema() pour obtenir le schéma de la table, puis le transmettre à setSchema().

Exemple

Le programme MapReduce suivant lit les données d'une table qu'il suppose avoir un entier dans la deuxième colonne ("colonne 1") et compte le nombre d'instances de chaque valeur distincte qu'il trouve. Autrement dit, il fait l'équivalent de "select col1, count(*) from $table group by col1;".

Par exemple, si les valeurs de la deuxième colonne sont {1, 1, 1, 3, 3, 5}, le programme produira la sortie suivante de valeurs et de décomptes -

1, 3
3, 2
5, 1

Regardons maintenant le code du programme -

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);
   }
}

Avant de compiler le programme ci-dessus, vous devez télécharger quelques jars et ajoutez-les au classpathpour cette application. Vous devez télécharger tous les bocaux Hive et 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).

Utilisez les commandes suivantes pour les copier jar fichiers de local à HDFS et ajoutez-les au 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

Utilisez la commande suivante pour compiler et exécuter le programme donné.

$HADOOP_HOME/bin/hadoop jar GroupByAge tmp/hive

Maintenant, vérifiez votre répertoire de sortie (hdfs: user / tmp / hive) pour la sortie (part_0000, part_0001).