Para replicar (CDC) dados para o Hive via Goldengate, neste exemplo irei utilizar os processos de extract e pump.

Como exemplo vou replicar (CDC) de uma tabela XPTO. Estou a mencionar o CDC (Change Data Capture), pois para o carregamento inicial, irei exemplificar com outra ferramenta que é o Sqoop.

Para a configuração do pump, será necessário configurar um adaptador Java, para realizar a integração com o Hive (Hive handler property file + Hive Handler Code).

Tendo como princípio, que temos a base de dados e o Goldengate devidamente instalados e prontos, podemos tratar da configuração.

1) Primeiramente, vamos ao nosso processo de extract (que vou chamar de ehive):

a) Adicionar Supplemental Logging:

ggsci>dblogin userid gguser, password xyz

ggsci>add trandata foo.cust_info, cols ( name, gender, city, phone, id, zip, cust_date)

Configure Oracle GoldenGate capture process.
Sample capture parameter file listed below

b) Parâmetros do processo ehive:

EXTRACT ehive
USERID gguser, PASSWORD xyz
EXTTRAIL ./dirdat/tf
GETUPDATEBEFORES
NOCOMPRESSUPDATES
TABLE foo.cust_info;

c) Adicionar o processo ehive:

ggsci>add extract ehive, tranlog, begin now
ggsci>add exttrail ./dirdat/tf, extract ehive, megabytes 50
ggsci>start extract ehive
ggsci>info ehive

2) Adicionar o processo de pump (que irei chamar de phive):

a) Parâmetros do processo phive:

extract phive
sourcedefs ./dirdef/efile.def
CUSEREXIT ./libggjava_ue.so CUSEREXIT PASSTHRU INCLUDEUPDATEBEFORES
TABLE foo.cust_info;

PS: É necessário gerar o mapa de estrutura de dados da tabela para permitir o funcionamento correto (SOURCEDEFS)

ggsci > Edit params defgen
defsfile ./dirdef/efile.def
userid ogg password ogg
TABLE foo.cust_info;

ggsci > sh ./defgen paramfile ./dirprm/defgen.prm

b) Adicionar o processo de pump phive:

ggsci> add extract phive, exttrailsource ./dirdat/tf

c) Configurar Hive handler property file:

Para isso, temos que criar o ficheiro/arquivo phive.properties em ./dirprm/:

vi ./dirprm/phive.properties

Conteúdo do phive.properties:

#Adapter Logging parameters.
gg.log=log4j
gg.log.level=info

#Adapter Check pointing parameters
goldengate.userexit.chkptprefix=HIVECHKP_
goldengate.userexit.nochkpt=true
# Java User Exit Property
goldengate.userexit.writers=jvm
jvm.bootoptions= -Xms64m -Xmx512M –Djava.class.path=ggjava/ggjava.jar:dirprm:dirprm/myhivehandler.jar:/home/hadoop/hadoop-0.20.203.0/hadoop-core-0.20.203.0.jar:/home/hadoop/hadoop-0.20.203.0/lib/commons-configuration-1.6.jar:/home/hadoop/hadoop-0.20.203.0/lib/commons-logging-api-1.0.4.jar:/home/hadoop/hadoop-0.20.203.0/lib/commons-lang-2.4.jar:/home/hadoop/hadoop-0.20.203.0:/home/hadoop/hadoop-0.20.203.0/conf:
#Properties for reporting statistics
# Minimum number of {records, seconds} before generating a report
jvm.stats.time=3600
jvm.stats.numrecs=5000
jvm.stats.display=TRUE
jvm.stats.full=TRUE

#Hive Handler.
gg.handlerlist=hivehandler
gg.handler.hivehandler.type=com.mycompany.bigdata.SampleHandlerHive
gg.handler.hivehandler.HDFSFileName=/user/hive/warehouse/xpto
gg.handler.hivehandler.RegularFileName=cinfo_hive.txt
gg.handler.hivehandler.RecordDelimiter=;
gg.handler.hivehandler.mode=tx

d) Agora é necessário compilar o adaptador Java (que pode ser customizado, mas aqui está o exemplo que utilizei):

Compilar o adaptador:

export CLASSPATH=.:../ggjava/ggjava.jar:dirprm:dirprm/myhdfshandler.jar:/opt/hadoop/hive/hcatalog/share/webhcat/svr/lib/hadoop-core-1.0.3.jar:/opt/hadoop/hive/lib/commons-configuration-1.6.jar:/opt/hadoop/hive/lib/commons-logging-api-1.0.4.jar:/opt/hadoop/hive/lib/commons-lang-2.4.jar:/opt/hadoop/hive:/opt/hadoop/hive/conf:/opt/hadoop:/opt/hadoop/conf

javac -d . SampleHiveHandler.java

Gerar o JAR file:

jar cvf myhivehandler.jar com

e) Iniciar o processo de pump phive:

ggsci>start phive
ggsci>info phive

3) Hora do teste 🙂

a) Inserts:

b) Status dos processos Goldengate:

c) Log do processo pump:

d) Dados no Hive

Para já, é isso, o proxímo post deste tema, será o carregamento inicial com o Sqoop.