HCatalog - Reader Writer

HCatalog contém uma API de transferência de dados para entrada e saída paralelas sem usar MapReduce. Esta API usa uma abstração de armazenamento básico de tabelas e linhas para ler dados do cluster Hadoop e gravar dados nele.

A API de transferência de dados contém principalmente três classes; aqueles são -

HCatReader - Lê dados de um cluster Hadoop.
HCatWriter - Grava dados em um cluster Hadoop.
DataTransferFactory - Gera instâncias de leitor e gravador.

Esta API é adequada para configuração de nó mestre-escravo. Vamos discutir mais sobreHCatReader e HCatWriter.

HCatReader

HCatReader é uma classe abstrata interna do HCatalog e abstrai as complexidades do sistema subjacente de onde os registros devem ser recuperados.

Sr. Não.	Nome e descrição do método
1	Public abstract ReaderContext prepareRead() throws HCatException Deve ser chamado no nó mestre para obter o ReaderContext, que deve ser serializado e enviado aos nós escravos.
2	Public abstract Iterator <HCatRecorder> read() throws HCaException Isso deve ser chamado nos nós escravos para ler HCatRecords.
3	Public Configuration getConf() Ele retornará o objeto da classe de configuração.

A classe HCatReader é usada para ler os dados do HDFS. A leitura é um processo de duas etapas em que a primeira ocorre no nó mestre de um sistema externo. A segunda etapa é realizada em paralelo em vários nós escravos.

As leituras são feitas em um ReadEntity. Antes de começar a ler, você precisa definir um ReadEntity para ler. Isso pode ser feito através deReadEntity.Builder. Você pode especificar um nome de banco de dados, nome de tabela, partição e string de filtro. Por exemplo -

ReadEntity.Builder builder = new ReadEntity.Builder();
ReadEntity entity = builder.withDatabase("mydb").withTable("mytbl").build(); 10.

O trecho de código acima define um objeto ReadEntity ("entidade"), que compreende uma tabela chamada mytbl em um banco de dados chamado mydb, que pode ser usado para ler todas as linhas desta tabela. Observe que esta tabela deve existir no HCatalog antes do início desta operação.

Depois de definir um ReadEntity, você obtém uma instância de HCatReader usando o ReadEntity e a configuração de cluster -

HCatReader reader = DataTransferFactory.getHCatReader(entity, config);

A próxima etapa é obter um ReaderContext do leitor da seguinte maneira -

ReaderContext cntxt = reader.prepareRead();

HCatWriter

Esta abstração é interna ao HCatalog. Isso é para facilitar a gravação no HCatalog de sistemas externos. Não tente instanciar isso diretamente. Em vez disso, use DataTransferFactory.

Sr. Não.	Nome e descrição do método
1	Public abstract WriterContext prepareRead() throws HCatException O sistema externo deve invocar este método exatamente uma vez a partir de um nó mestre. Retorna umWriterContext. Isso deve ser serializado e enviado aos nós escravos para construirHCatWriter há.
2	Public abstract void write(Iterator<HCatRecord> recordItr) throws HCaException Este método deve ser usado em nós escravos para realizar gravações. O recordItr é um objeto iterador que contém a coleção de registros a serem gravados no HCatalog.
3	Public abstract void abort(WriterContext cntxt) throws HCatException Este método deve ser chamado no nó mestre. O objetivo principal deste método é fazer limpezas em caso de falhas.
4	public abstract void commit(WriterContext cntxt) throws HCatException Este método deve ser chamado no nó mestre. O objetivo deste método é fazer commit de metadados.

Semelhante à leitura, a escrita também é um processo de duas etapas em que a primeira etapa ocorre no nó mestre. Posteriormente, a segunda etapa ocorre em paralelo nos nós escravos.

As gravações são feitas em um WriteEntity que pode ser construído de forma semelhante à leitura -

WriteEntity.Builder builder = new WriteEntity.Builder();
WriteEntity entity = builder.withDatabase("mydb").withTable("mytbl").build();

O código acima cria um objeto WriteEntity entityque pode ser usado para escrever em uma tabela chamadamytbl no banco de dados mydb.

Depois de criar um WriteEntity, a próxima etapa é obter um WriterContext -

HCatWriter writer = DataTransferFactory.getHCatWriter(entity, config);
WriterContext info = writer.prepareWrite();

Todas as etapas acima ocorrem no nó mestre. O nó mestre então serializa o objeto WriterContext e o disponibiliza para todos os escravos.

Em nós escravos, você precisa obter um HCatWriter usando WriterContext da seguinte forma -

HCatWriter writer = DataTransferFactory.getHCatWriter(context);

Então o writerleva um iterador como argumento para o writemétodo -

writer.write(hCatRecordItr);

o writer então liga getNext() neste iterador em um loop e grava todos os registros anexados ao iterador.

o TestReaderWriter.javaarquivo é usado para testar as classes HCatreader e HCatWriter. O programa a seguir demonstra como usar HCatReader e HCatWriter API para ler dados de um arquivo de origem e, subsequentemente, gravá-los em um arquivo de destino.

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.ql.CommandNeedRetryException;
import org.apache.hadoop.mapreduce.InputSplit;

import org.apache.hive.HCatalog.common.HCatException;
import org.apache.hive.HCatalog.data.transfer.DataTransferFactory;
import org.apache.hive.HCatalog.data.transfer.HCatReader;
import org.apache.hive.HCatalog.data.transfer.HCatWriter;
import org.apache.hive.HCatalog.data.transfer.ReadEntity;
import org.apache.hive.HCatalog.data.transfer.ReaderContext;
import org.apache.hive.HCatalog.data.transfer.WriteEntity;
import org.apache.hive.HCatalog.data.transfer.WriterContext;
import org.apache.hive.HCatalog.mapreduce.HCatBaseTest;

import org.junit.Assert;
import org.junit.Test;

public class TestReaderWriter extends HCatBaseTest {
   @Test
   public void test() throws MetaException, CommandNeedRetryException,
      IOException, ClassNotFoundException {
		
      driver.run("drop table mytbl");
      driver.run("create table mytbl (a string, b int)");
		
      Iterator<Entry<String, String>> itr = hiveConf.iterator();
      Map<String, String> map = new HashMap<String, String>();
		
      while (itr.hasNext()) {
         Entry<String, String> kv = itr.next();
         map.put(kv.getKey(), kv.getValue());
      }
		
      WriterContext cntxt = runsInMaster(map);
      File writeCntxtFile = File.createTempFile("hcat-write", "temp");
      writeCntxtFile.deleteOnExit();
		
      // Serialize context.
      ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(writeCntxtFile));
      oos.writeObject(cntxt);
      oos.flush();
      oos.close();
		
      // Now, deserialize it.
      ObjectInputStream ois = new ObjectInputStream(new FileInputStream(writeCntxtFile));
      cntxt = (WriterContext) ois.readObject();
      ois.close();
      runsInSlave(cntxt);
      commit(map, true, cntxt);
		
      ReaderContext readCntxt = runsInMaster(map, false);
      File readCntxtFile = File.createTempFile("hcat-read", "temp");
      readCntxtFile.deleteOnExit();
      oos = new ObjectOutputStream(new FileOutputStream(readCntxtFile));
      oos.writeObject(readCntxt);
      oos.flush();
      oos.close();
		
      ois = new ObjectInputStream(new FileInputStream(readCntxtFile));
      readCntxt = (ReaderContext) ois.readObject();
      ois.close();
		
      for (int i = 0; i < readCntxt.numSplits(); i++) {
         runsInSlave(readCntxt, i);
      }
   }
	
   private WriterContext runsInMaster(Map<String, String> config) throws HCatException {
      WriteEntity.Builder builder = new WriteEntity.Builder();
      WriteEntity entity = builder.withTable("mytbl").build();
		
      HCatWriter writer = DataTransferFactory.getHCatWriter(entity, config);
      WriterContext info = writer.prepareWrite();
      return info;
   }
	
   private ReaderContext runsInMaster(Map<String, String> config, 
      boolean bogus) throws HCatException {
      ReadEntity entity = new ReadEntity.Builder().withTable("mytbl").build();
      HCatReader reader = DataTransferFactory.getHCatReader(entity, config);
      ReaderContext cntxt = reader.prepareRead();
      return cntxt;
   }
	
   private void runsInSlave(ReaderContext cntxt, int slaveNum) throws HCatException {
      HCatReader reader = DataTransferFactory.getHCatReader(cntxt, slaveNum);
      Iterator<HCatRecord> itr = reader.read();
      int i = 1;
		
      while (itr.hasNext()) {
         HCatRecord read = itr.next();
         HCatRecord written = getRecord(i++);
			
         // Argh, HCatRecord doesnt implement equals()
         Assert.assertTrue("Read: " + read.get(0) + "Written: " + written.get(0),
         written.get(0).equals(read.get(0)));
			
         Assert.assertTrue("Read: " + read.get(1) + "Written: " + written.get(1),
         written.get(1).equals(read.get(1)));
			
         Assert.assertEquals(2, read.size());
      }
		
      //Assert.assertFalse(itr.hasNext());
   }
	
   private void runsInSlave(WriterContext context) throws HCatException {
      HCatWriter writer = DataTransferFactory.getHCatWriter(context);
      writer.write(new HCatRecordItr());
   }
	
   private void commit(Map<String, String> config, boolean status,
      WriterContext context) throws IOException {
      WriteEntity.Builder builder = new WriteEntity.Builder();
      WriteEntity entity = builder.withTable("mytbl").build();
      HCatWriter writer = DataTransferFactory.getHCatWriter(entity, config);
		
      if (status) {
         writer.commit(context);
      } else {
         writer.abort(context);
      }
   }
	
   private static HCatRecord getRecord(int i) {
      List<Object> list = new ArrayList<Object>(2);
      list.add("Row #: " + i);
      list.add(i);
      return new DefaultHCatRecord(list);
   }
	
   private static class HCatRecordItr implements Iterator<HCatRecord> {
      int i = 0;
		
      @Override
      public boolean hasNext() {
         return i++ < 100 ? true : false;
      }
		
      @Override
      public HCatRecord next() {
         return getRecord(i);
      }
		
      @Override
      public void remove() {
         throw new RuntimeException();
      }
   }
}

O programa acima lê os dados do HDFS na forma de registros e grava os dados do registro em mytable

Next page ↱