De MariaDB a HDFS: Usando Continuent Tungsten. Parte 1

Semana santa y yo con nuevas batallas que contar.
Me hayaba yo en el trabajo, pensando en que iba a invertir la calma que acompa;a a los dias de vacaciones que libremente podemos elegir trabajar y pense: No seria bueno terminar esa sincronizacion entre los servidores de mariaDB y HIVE?

Ya habia buscado algo de info al respecto en Enero hasta tenia una PoC montada con unas VM que volvi a encender, pero estaba todo podrido: no arrancaba, no funcionba ni siquiera me acordaba como lo habia hecho y el history de la shell er un galimatias. Decidi que si lo rehacia todo desde cero iba a poder dejarlo escrito en un playbook y ademas, aprenderlo y automatizarlo hasta el limite de poder desplegar de forma automatica on Ansible.

El origen

El problema viene de los Devilopers que tenemos, necesitaban una forma de pasar los datos entre los entornos de middleware. Todo lo meten a Kafka y tienen jobs que copian los datos entre microservicios, pero para tener toda la info en Hadoop desde los diferentes entornos (Dev, Staging, QA), teniamos una solucion de Sqoop que era antidiluviana: dump y restore de toda la base de datos. Y esto es, porque sqoop no permite per se hacer una carga incremental, necesita que le digas a partir de que fila quieres que siga importando, es decir:

  1. El lunes se importa de la fila 1 a la 100
  2. el martes de la 101 a la 500
  3. miercoles se actualiza una fila de el primer rango
  4. El Jueves tienes un problema cuando un QA te dice que los datos estan mal.

Asi que nosotros, como eran entornos pequenos y estabamos en mentalidad PoC, pues haciamos dump de las tablas y luego con Sqoop, re-importabamos todo. Claro, por el momento vale, ahora cuando falla el job, tenemos a varios dandonos toques en la espalda en plan que no pueden trabajar porque los datos son viejos.
Total, que buscando soluciones, aparecio Continuent Tungsten Replicator…ohhhhh!

Nuestra solucion

Este replicador fue desarrollado en principio por VMware y luego ha ido volando solo (no se si aun es parte de VMware o no), teniendo dos vertientes, la corporate y la opensource. Yo me estoy refiriendo siempre a la OpenSource (OSS la llaman ellos).
Este replicador funciona en java y te lo bajas del git, lo compilas siguiendo las instrucciones y lo ejecutas en las maquinas que vayan a ser miembros de los canales, definiendo quien va a ser master y quien slave.

Los canales son canales de replica, en nuestro caso, tendriamos uno por cada entorno (dev, qa, staging, pro, etc), que los vamos a llamar como los entornos o podeis usar la nomenclatura que usan ellos: (alpha, beta, gamma, etc).

El replicador funciona de la siguiente manera, en nuestro caso que lo que queremos es replicar de MariaDB a HDFS:

  • El source lee del binlog la transaccion de mariaDB
  • Le asigna un id unico y lo guarda en una bd de datos, que esta definido por cada canal y que en este caso se llama tungsten_alpha
  • El destino, se conecta al source y se trae los datos
  • Este lo escribe en la carpeta de HDFS del usuario tungsten en el directorio de staging
  • Un job en el destino procesa este csv y los convierte a uno que sea entendible por HIVE

Estos son los diagramas de como funciona la replicacion (hotlinkeado desde su web)

Y esto como transforma los CSV Stage Data a CSV Active Data para ser interpretados por Hive como datos, tablas y bases de datos.

Una vez teniendo claro como funciona el tema, la arquitectura que he desplegado es la siguiente, cada entorno vuelca sus mariaDBs a el datalake de Hadoop usando canales. Cada canal, es un entorno y en el datalake, tengo tantos listeners como replicators en los origenes.

Lo primero, creamos los pre-requisitos del tungsten usando un playbook de ansible, que cuando lo tenga listo, lo subire al Ansible Galaxy, pero debe de tener esto:

  • creamos el usuario local tungsten y lo agregamos al grupo de hdfs y al de wheel.
  • instalamos las dependencias si queremos hacer el build del tungsten en cada maquina o usamos el tarball (opcional)
  • creamos el usuario de tungsten en el HDFS y lo hacemos propietario de su carpeta y mode 700
  • hacemos que el usuario tungsten pueda acceder a los servidores entre ellos usando clave ssh

Luego, preparamos los hosts (el de origen que es el de mariaDB y el de destino, un nodo de HDFS).

Por lo que mas querais, preparad bien los entornos, paso por paso. Os va a ahorrar mucho debugging

Preparar el host de mariaDB

APUNTE
Yo para tener los hosts de mariaDB sincronizado pero sin carga, los tengo metidos en el cluster pero el proxy (Maxscale) que esta por delante, no sabe de la existencia del mismo, asi que tengo un nodo solo para la replica de datos y de apoyo al Quorum, a diferencia de los dos nodos activos, que son un multi-master fisico de 64 cores y 128 GB solo con SSDs en raid 10.

Pasos a configurar en vuestro my.cnf

1
2
3
4
binlog-format = row
character-set-server=utf8
collation-server=utf8_general_ci
default-time-zone='+00:00'

Y si es > MySQL 5.6.2
binlog-row-image = full

Y en el server de SQL
mysql> SET GLOBAL binlog-format = ROW;

Preparar el host de Hadoop (HDFS)

DOCUMENTACION
Toda la documentacion oficial de como preparar los hosts, la instalacion del replicator y demas, la teneis en estos links:
MySQL > aqui
Hadoop > aqui

Arrancar el servicio

Una vez que:

  • Habeis definido la arquitectura
  • Teneis el replicator instalado en todos los Hosts
  • Esta correctamente configurado el firewall y los hostsnames
  • Teneis un cafe humeante en la mano y los Lusers estan ocupados lidiando con el excel

Podeis, instalar el servicio ejecutando lo siguiente desde la ruta donde habeis instalado el replicator (en mi maquina es esta):
/opt/tungsten-replicator/builder/build/tungsten-replicator-5.2.1/tools

** MariaDB host(s) **

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# Options for the alpha data service
$:./tpm install alpha \
--master=master-dev \
--install-directory=/opt/continuent \
--replication-user=tungsten \
--replication-password=##PASSWORD## \
--mysql-allow-intensive-checks=true \
--mysql-enable-enumtostring=true \
--mysql-enable-settostring=true \
--mysql-use-bytes-for-string=false \
--property=replicator.filter.pkey.addColumnsToDeletes=true \
--property=replicator.filter.pkey.addPkeyToInserts=true \
--enable-batch-service=true \
--skip-validation-check=HostsFileCheck \
--skip-validation-check=MySQLMyISAMCheck \
--skip-validation-check=ReplicationServicePipelines \
--svc-extractor-filters=colnames,pkey
--start-and-report=true

Hadoop host(s)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
$:./tools/tpm install alpha \
--batch-enabled=true \
--batch-load-language=js \
--batch-load-template=hadoop \
--datasource-type=file \
--install-directory=/opt/continuent \
--master=host1 \
--members=host2 \
--property=replicator.datasource.global.csvType=hive \
--property=replicator.stage.q-to-dbms.blockCommitInterval=1s \
--property=replicator.stage.q-to-dbms.blockCommitRowCount=1000 \
--replication-password=seen%GAIN%DISH%12 \
--replication-user=tungsten \
--skip-validation-check=DatasourceDBPort \
--skip-validation-check=DirectDatasourceDBPort \
--skip-validation-check=HostsFileCheck \
--skip-validation-check=InstallerMasterSlaveCheck \
--skip-validation-check=ReplicationServicePipelines \
--rmi-port=25550 \
--start-and-report=true

El ejecutarlo como root nos creara una carpeta /opt/continuent con los datos de la replicacion, el jar y todo lo necesario para luego arrancar el replicator

MariaDB

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
$:./tpm install alpha   
--master=master-agp
--install-directory=/opt/continuent
--replication-user=tungsten
--replication-password={{password}}
--mysql-allow-intensive-checks=true
--mysql-enable-enumtostring=true
--mysql-enable-settostring=true
--mysql-use-bytes-for-string=false
--property=replicator.filter.pkey.addColumnsToDeletes=true --property=replicator.filter.pkey.addPkeyToInserts=true
--enable-batch-service=true
--skip-validation-check=HostsFileCheck --skip-validation-check=MySQLMyISAMCheck
--skip-validation-check=ReplicationServicePipelines   
--svc-extractor-filters=colnames,pkey
--start-and-report=true

NOTE  >> Data service(s) alpha updated in   /opt/deleteme/tungsten-replicator/builder/build/tungsten-replicator-5.2.1/deploy.cfg

WARN  >> master_agp >> ntpd is not running. It is important that configured hosts have time synchronised. (NtpdRunningCheck)  
WARN  >> master_agp >> "sync_binlog" is set to 0 in the MySQL configuration file for tungsten@master-agp:3306 (WITH PASSWORD) this setting can lead to possible data loss in a server failure (MySQLSettingsCheck)  
WARN  >> master_agp >> We suggest adding "max_allowed_packet=52m" or greater to the MySQL configuration file for tungsten@master-agp:3306 (WITH PASSWORD) (MySQLSettingsCheck)  
WARN  >> master_agp >> We suggest adding "open_files_limit=65535" to the MySQL configuration file for tungsten@master-agp:3306 (WITH PASSWORD) (MySQLSettingsCheck)  
WARN  >> master_agp >> Add '*       -    nofile  65535' to your /etc/security/limits.conf and restart MySQL to make sure the setting takes effect (MySQLSettingsCheck)  

#####################################################################  
# Next Steps  
#####################################################################  
Once your services start successfully replication will begin.
To look at services and perform administration, run the following command
from any database server.

  /opt/continuent/tungsten/tungsten-replicator/bin/trepctl services

Configuration is now complete.  For further information, please consult
Tungsten documentation, which is available at docs.continuent.com.  

NOTE  >> Command successfully completed

Hadoop

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
$:/opt/tungsten/tungsten-replicator-5.2.1/tools$ ./tpm update alpha --batch-enabled=true
--batch-load-language=js
--batch-load-template=hadoop
--datasource-type=file
--install-directory=/opt/continuent
--master=master-agp
--members=hdp-dw-1-dn-1.domain.tld
--property=replicator.datasource.global.csvType=hive --property=replicator.stage.q-to-dbms.blockCommitInterval=1s --property=replicator.stage.q-to-dbms.blockCommitRowCount=1000
--replication-password=seen%GAIN%DISH%12
--replication-user=tungsten
--skip-validation-check=DatasourceDBPort
--skip-validation-check=DirectDatasourceDBPort
--skip-validation-check=HostsFileCheck
--skip-validation-check=InstallerMasterSlaveCheck
--skip-validation-check=ReplicationServicePipelines
--rmi-port=25550 --start-and-report=true

NOTE  >> Log output being sent to /tmp/tungsten/tungsten-configure.log
NOTE  >> Data service(s) alpha updated in /opt/tungsten/tungsten-replicator-5.2.1/deploy.cfg  

WARN  >> hdp_dw_1_dn_1_domain_local >> The open file limit is set to '1024', we suggest a value of 65535. Add '*       -    nofile  65535' to your /etc/security/limits.conf and restart your session (OpenFilesLimitCheck)  
WARN  >> hdp_dw_1_dn_1_domain_local >> ntpd is not running. It is important that configured hosts have time synchronised. (NtpdRunningCheck)  

#####################################################################
# Next Steps
#####################################################################
Once your services start successfully replication will begin.
To look at services and perform administration, run the following command
from any database server.

  /opt/continuent/tungsten/tungsten-replicator/bin/trepctl services

Configuration is now complete.  For further information, please consult
Tungsten documentation, which is available at docs.continuent.com.

NOTE  >> Command successfully completed

Nos vamos a la carpta que nos habra creado en /opt/continuent/tungsten/tunsgten-replicator/bin en el master (que sera el de SQL) y ejecutamos ./trepctl status

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
Processing status command...
NAME                     VALUE
----                     -----
appliedLastEventId     : master-binlog.000021:0000000000715329;-1
appliedLastSeqno       : 1741
appliedLatency         : 0.526
autoRecoveryEnabled    : false
autoRecoveryTotal      : 0
channels               : 1
clusterName            : alpha
currentEventId         : master-binlog.000021:0000000000715329
currentTimeMillis      : 1522410957470
dataServerHost         : master-agp
extensions             :
host                   : master-agp
latestEpochNumber      : 0
masterConnectUri       : thl://localhost:/
masterListenUri        : thl://master-agp:2112/
maximumStoredSeqNo     : 1741
minimumStoredSeqNo     : 0
offlineRequests        : NONE
pendingError           : NONE
pendingErrorCode       : NONE
pendingErrorEventId    : NONE
pendingErrorSeqno      : -1
pendingExceptionMessage: NONE
pipelineSource         : jdbc:mysql:thin://master-agp:3306/tungsten_alpha?noPrepStmtCache=true
relativeLatency        : 1.47
resourcePrecedence     : 99
rmiPort                : 10000
role                   : master
seqnoType              : java.lang.Long
serviceName            : alpha
serviceType            : local
simpleServiceName      : alpha
siteName               : default
sourceId               : master-agp
state                  : ONLINE
timeInStateSeconds     : 1270.349
timezone               : GMT
transitioningTo        :
uptimeSeconds          : 1271.08
useSSLConnection       : false
version                : Tungsten Replicator 5.2.1
Finished status command...

Antes de ejecutarlo en el hadoop, tenemos que activar el servicio

1
2
3
$:/opt/continuent/tungsten/tungsten-replicator/bin$ ./replicator start
Waiting for Tungsten Replicator Service.
running: PID:15600

Y ahora ejecutamos

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
$:/opt/continuent/tungsten/tungsten-replicator/bin$ ./trepctl status
Processing status command...
NAME                     VALUE
----                     -----
appliedLastEventId     : NONE
appliedLastSeqno       : -1
appliedLatency         : -1.0
autoRecoveryEnabled    : false
autoRecoveryTotal      : 0
channels               : 1
clusterName            : alpha
currentEventId         : NONE
currentTimeMillis      : 1522411863421
dataServerHost         : hdp-dw-1-dn-1.domain.tld
extensions             :
host                   : hdp-dw-1-dn-1.domain.tld
latestEpochNumber      : -1
masterConnectUri       : thl://master-agp:2112/
masterListenUri        : null
maximumStoredSeqNo     : -1
minimumStoredSeqNo     : -1
offlineRequests        : NONE
pendingError           : NONE
pendingErrorCode       : NONE
pendingErrorEventId    : NONE
pendingErrorSeqno      : -1
pendingExceptionMessage: NONE
pipelineSource         : UNKNOWN
relativeLatency        : -1.0
resourcePrecedence     : 99
rmiPort                : 25550
role                   : slave
seqnoType              : java.lang.Long
serviceName            : alpha
serviceType            : local
simpleServiceName      : alpha
siteName               : default
sourceId               : hdp-dw-1-dn-1.domain.tld
state                  : GOING-ONLINE:SYNCHRONIZING
timeInStateSeconds     : 2.972
timezone               : GMT
transitioningTo        :
uptimeSeconds          : 5.42
useSSLConnection       : false
version                : Tungsten Replicator 5.2.1
Finished status command...

El cual en nuestro caso, no va a funcionar porque no tenemos el puerto abierto, 2112 a traves del firewall. Como veis aqui masterConnectUri : thl://master-agp:2112/ intenta conectar usando el fichero definido en los /etc/hosts
Asi que lo habilitamos en el firewall y….

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
Processing status command...
NAME                     VALUE
----                     -----
appliedLastEventId     : master-binlog.000021:0000000000000629;-1
appliedLastSeqno       : 0
appliedLatency         : 2816.186
autoRecoveryEnabled    : false
autoRecoveryTotal      : 0
channels               : 1
clusterName            : alpha
currentEventId         : NONE
currentTimeMillis      : 1522412558674
dataServerHost         : hdp-dw-1-dn-1.domain.tld
extensions             :
host                   : hdp-dw-1-dn-1.domain.tld
latestEpochNumber      : 0
masterConnectUri       : thl://master-agp:2112/
masterListenUri        : null
maximumStoredSeqNo     : 3861
minimumStoredSeqNo     : 0
offlineRequests        : NONE
pendingError           : NONE
pendingErrorCode       : NONE
pendingErrorEventId    : NONE
pendingErrorSeqno      : -1
pendingExceptionMessage: NONE
pipelineSource         : thl://master-agp:2112/
relativeLatency        : 2871.674
resourcePrecedence     : 99
rmiPort                : 25550
role                   : slave
seqnoType              : java.lang.Long
serviceName            : alpha
serviceType            : local
simpleServiceName      : alpha
siteName               : default
sourceId               : hdp-dw-1-dn-1.domain.tld
state                  : ONLINE
timeInStateSeconds     : 55.649
timezone               : GMT
transitioningTo        :
uptimeSeconds          : 700.673
useSSLConnection       : false
version                : Tungsten Replicator 5.2.1
Finished status command...

Tenemos conexion!!!!
Ahora solo nos queda la segunda parte, que es la de configurar el Batch Process para convertir los staging CSV a Data Active CSV.

Pero eso otro dia.

Links
[1] continuent 5.3 tungsten replicator

Comments

⬆︎TOP