Page Comparison

Table of Contents

This document lists the detailed steps required for setting up Replication for CDAP :replication.

Cluster Setup

CDAP replication relies on the cluster administrator setting up replication on HBase, HDFS, Hive, and Kafka. It is assumed that CDAP is only running on the master cluster. It is assumed that you have not started CDAP before any of these steps.

HBase

Install the relevant cdap-hbase-compat package package on all hbase HBase nodes in your cluster . Compat packages are:
cdap-hbase-compat-0.96

cdap-hbase-compat-0.98

in order to use the replication status coprocessors. Note that due to HBase limitations, these coprocessors cannot be used on HBase 0.96 or 0.98. Available compat packages are:

cdap-hbase-compat-1.0
cdap-hbase-compat-1.0-cdh
cdap-hbase-compat-1.0-cdh5.5.0
cdap-hbase-compat-1.1
cdap-hbase-compat-1.2-cdh5.7.0

Modify hbase-site.xml on on all hbase HBase nodes to enable hbase replicationHBase replication, and to use the CDAP replication status coprocessors:

Code Block

<property>
  <name>hbase.replication</name>
  <value>true</value>
</property>
<property>
  <name>hbase.coprocessor.regionserver.classes</name>
  <value>co.cask.cdap.data2.replication.LastReplicateTimeObserver</value>
</property>
<property>
  <name>hbase.coprocessor.wal.classes</name>
  <value>co.cask.cdap.data2.replication.LastWriteTimeObserver</value>
</property>

Modify hbase-env.sh on on all hbase nodes HBase nodes to include the hbase coprocessor HBase coprocessor in the classpath classpath:

Code Block

export HBASE_CLASSPATH="$HBASE_CLASSPATH:/opt/cdap/<hbase-compat-version>/coprocessor/*"
 
# for example, if you're on cdh5.5.x and have installed the cdap-hbase-compat-1.0-cdh5.5.0 package:
export HBASE_CLASSPATH="$HBASE_CLASSPATH:/opt/cdap/hbase-compat-1.0-cdh5.5.0/coprocessor/*"

Restart hbase master HBase master and regionservers

HDFS

Setup HDFS replication using the solution provided by your distribution. HDFS does not have true replication, but is usually achieved by scheduling regular distcp jobs.

Hive

Setup replication for the database backing your Hive Metastore. Note that this will simply replicate the Hive metadata (which tables exist, table metadata, etc.), but not the data itself. It is assumed you will not be running Hive queries on the slave until a manual failover occurs.

For example, to setup MySQL replication, follow the steps at https://dev.mysql.com/doc/refman/5.7/en/replication-howto.html, which amount to:

...

Modify my.cnf on the master to set a server-id and use bin logging

Code Block
[mysqld] log-bin=mysql-bin server-id=1

...

Restart mysql on master

...

Modify my.cnf on the slave to set a server-id

Code Block
[mysqld] server-id=2

...

Restart mysql on slave

...

Create a replication user on the master

Code Block
mysql> CREATE USER 'repl'@'%.mydomain.com' IDENTIFIED BY 'slavepass'; mysql> GRANT REPLICATION SLAVE ON . TO 'repl'@'%.mydomain.com';

...

Obtain the master status

Code Block

mysql > SHOW MASTER STATUS;
+------------------+----------+--------------+------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+------------------+----------+--------------+------------------+
| mysql-bin.000003 | 73       | test         | manual,mysql     |
+------------------+----------+--------------+------------------+

...

Set master on the slave

Code Block

mysql> CHANGE MASTER TO
    ->     MASTER_HOST='master_host_name',
    ->     MASTER_USER='replication_user_name', // repl
    ->     MASTER_PASSWORD='replication_password', //slavepass
    ->     MASTER_LOG_FILE='recorded_log_file_name', //mysql-bin.000003
    ->     MASTER_LOG_POS=recorded_log_position; // 73

...

Start slave

Code Block
mysql> start slave

Verify slave status

...

Enable replication from master to slave:

Code Block
master hbase shell> add_peer '[slave-name]', '[slave-zookeeper-quorum]:/[slave-zk-node]' example: master hbase shell> add_peer 'hbase2', 'hbase2-master.domain.net:2181:/hbase'

Enable replication from slave to master:

Code Block
slave hbase shell> add_peer '[master-name]', '[master-zookeeper-quorum]:/[master-zk-node]' example: slave hbase shell> add_peer 'hbase1', 'hbase1-master.domain.net:2181:/hbase'

Confirm that replication is working:

Code Block

master hbase shell> create 'repltest', 'f'
slave hbase shell> create 'repltest', 'f'
master hbase shell> enable_table_replication 'repltest'
slave hbase shell> alter 'repltest', { 'NAME' => 'f', 'REPLICATION_SCOPE' => 1 }
master hbase shell> put 'repltest', 'masterrow', 'f:v1', 'v1'
slave hbase shell> put 'repltest', 'slaverow', 'f:v1', 'v1'
master hbase shell> scan 'repltest'
slave hbase shell> scan 'repltest'

HDFS

Set up HDFS replication using the solution provided by your distribution. HDFS does not have true replication, but it is usually achieved by scheduling regular distcp jobs.

Hive

Set up replication for the database backing your Hive Metastore. Note that this will simply replicate the Hive metadata (which tables exist, table metadata, etc.), but not the data itself. It is assumed you will not be running Hive queries on the slave until a manual failover occurs.

For example, to setup MySQL replication, follow the steps at https://dev.mysql.com/doc/refman/5.7/en/replication-howto.html, which amount to:

Modify my.cnf on the master to set a server-id and use bin logging:
Code Block
[mysqld] log-bin=mysql-bin binlog_format=row server-id=1
Restart mysql on the master
Modify my.cnf on the slave to set a server-id:
Code Block
[mysqld] server-id=2
Restart MySQL on the slave
Create a replication user on the master:
Code Block
mysql> CREATE USER 'repl'@'%' IDENTIFIED BY 'slavepass'; mysql> GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';
In a production setup, you probably want to restrict the hostname instead of using '%'. Depending on your MySQL version, you may need to use old_password:
Code Block
mysql> SET PASSWORD FOR 'repl'@'%' = OLD_PASSWORD('slavepass');

Obtain the master status:

Code Block

mysql > SHOW MASTER STATUS;
+------------------+----------+--------------+------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+------------------+----------+--------------+------------------+
| mysql-bin.000003 | 73       | test         | manual,mysql     |
+------------------+----------+--------------+------------------+

Set master on the slave:

Code Block

mysql> CHANGE MASTER TO
    ->     MASTER_HOST='master_host_name',
    ->     MASTER_USER='replication_user_name', // repl
    ->     MASTER_PASSWORD='replication_password', //slavepass
    ->     MASTER_LOG_FILE='recorded_log_file_name', //mysql-bin.000003
    ->     MASTER_LOG_POS=recorded_log_position; // 73

Start the slave:
Code Block
mysql> start slave

Verify slave status:

Code Block

mysql> SHOW SLAVE STATUS\G
*************************** 1. row ***************************
               Slave_IO_State: Waiting for master to send event
                  Master_Host: localhost
                  Master_User: root
                  Master_Port: 13000
                Connect_Retry: 60
              Master_Log_File: master-bin.000002
          Read_Master_Log_Pos: 1307
               Relay_Log_File: slave-relay-bin.000003
                Relay_Log_Pos: 1508
        Relay_Master_Log_File: master-bin.000002
             Slave_IO_Running: Yes
            Slave_SQL_Running: Yes
              Replicate_Do_DB:
          Replicate_Ignore_DB:
    Slave_IO_State: Waiting for master to send event Replicate_Do_Table:
       Replicate_Ignore_Table:
      Replicate_Wild_Do_Table:
  Master_HostReplicate_Wild_Ignore_Table:
localhost                   MasterLast_UserErrno: root0
                   MasterLast_PortError:
13000                 Connect_Retry: 60   Skip_Counter: 0
          Exec_Master_Log_FilePos: master-bin.000002 1307
              ReadRelay_Master_Log_PosSpace: 13071858
               Relay_Log_File: slave-relay-bin.000003Until_Condition: None
                RelayUntil_Log_PosFile: 1508
                Relay_MasterUntil_Log_File: master-bin.000002
Pos: 0
            SlaveMaster_IOSSL_RunningAllowed: YesNo
            Slave_SQL_RunningMaster_SSL_CA_File:
Yes           Master_SSL_CA_Path:
   Replicate_Do_DB:           ReplicateMaster_IgnoreSSL_DBCert:
           Replicate Master_DoSSL_TableCipher:
          Replicate_Ignore_Table:     Master_SSL_Key:
        Replicate_WildSeconds_DoBehind_TableMaster: 0
 Replicate_Wild_Ignore_Table:
Master_SSL_Verify_Server_Cert: No
                  Last_IO_Errno: 0
                   Last_IO_Error:
                 Skip_CounterLast_SQL_Errno: 0
          Exec_Master_Log_Pos: 1307
     Last_SQL_Error:
  Replicate_Ignore_Server_Ids:
             RelayMaster_Log_Space: 1858Server_Id: 1
                  UntilMaster_ConditionUUID: None3e11fa47-71ca-11e1-9e33-c80aa9429562
               UntilMaster_LogInfo_File: /var/mysqld.2/data/master.info
               Until_Log_Pos: 0            Master_SSL_AllowedSQL_Delay: No0
           MasterSQL_SSLRemaining_CA_FileDelay: NULL
          Master_SSL_CA_Path:
              Master_SSL_Cert:Slave_SQL_Running_State: Slave has read all relay log; waiting for the slave I/O thread to update it
            Master_SSLRetry_CipherCount: 10
                  Master_SSL_KeyBind:
        Seconds_Behind_Master: 0
Master_SSL_Verify_Server_Cert: NoLast_IO_Error_Timestamp:
     Last_SQL_Error_Timestamp:
                 LastMaster_IOSSL_ErrnoCrl:
0           Master_SSL_Crlpath:
      Last_IO_Error:
     Retrieved_Gtid_Set: 3e11fa47-71ca-11e1-9e33-c80aa9429562:1-5
              LastExecuted_SQL_Errno: 0Gtid_Set: 3e11fa47-71ca-11e1-9e33-c80aa9429562:1-5
                LastAuto_SQL_ErrorPosition:   Replicate_Ignore_Server_Ids:1
             Master_Server_Id: 1Replicate_Rewrite_DB:
                     Master_UUID: 3e11fa47-71ca-11e1-9e33-c80aa9429562
             Master_Info_File: /var/mysqld.2/data/master.info
                    SQL_Delay: 0
          SQL_Remaining_Delay: NULL
      Slave_SQL_Running_State: Slave has read all relay log; waiting for the slave I/O thread to update it
           Master_Retry_Count: 10
                  Master_Bind:
      Last_IO_Error_Timestamp:
     Last_SQL_Error_Timestamp:
               Master_SSL_Crl:
           Master_SSL_Crlpath:
           Retrieved_Gtid_Set: 3e11fa47-71ca-11e1-9e33-c80aa9429562:1-5
            Executed_Gtid_Set: 3e11fa47-71ca-11e1-9e33-c80aa9429562:1-5
                Auto_Position: 1
         Replicate_Rewrite_DB:
                 Channel_name:

Kafka

Setup replication for the Kafka brokers you are using. Kafka MirrorMaker is the most common solution.

CDAP Setup

CDAP requires that you provide an extension that will perform HBase related DDL operations on both clusters instead of just one. To create an extension, you must implement the HBaseDDLExecutor class:

...

Channel_name:

Kafka

Set up replication for the Kafka brokers you are using. Kafka MirrorMaker (see https://kafka.apache.org/documentation.html#basic_ops_mirror_maker and https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330) is the most common solution.

CDAP Setup

CDAP requires that you provide an extension that will perform HBase-related DDL operations on both clusters instead of on just one. To create an extension, you must implement the HBaseDDLExecutor class:

Code Block

/**
 * Interface providing the HBase DDL operations.
 */
public interface HBaseDDLExecutor extends Closeable {
  /**
   * Initialize the {@link HBaseDDLExecutor}.
   * @param context the context for the executor
   */
  void initialize(HBaseDDLExecutorContext context);

  /**
   * Create the specified namespace if it does not exist.
   *
   * @param name the namespace to create
   * @throws IOException if a remote or network exception occurs
   */
  boolean createNamespaceIfNotExists(String name) throws IOException;

  /**
   * Delete the specified namespace if it exists.
   *
   * @param name the namespace to delete
   * @throws IOException if a remote or network exception occurs
   * @throws IllegalStateException if there are tables in the namespace
   */
  void deleteNamespaceIfExists(String name) throws IOException;

  /**
   * Create the specified table if it does not exist.
   *
   * @param descriptor the descriptor for the table to create
   * @param splitKeys the initial split keys for the table
   * @throws IOException if a remote or network exception occurs
   */
  void createTableIfNotExists(TableDescriptor descriptor, @Nullable byte[][] splitKeys)
    throws IOException;

  /**
   * Enable the specified table if it is disabled.
   *
   * @param namespace the namespace of the table to enable
   * @param name the name of the table to enable
   * @throws IOException if a remote or network exception occurs
   */
  void createNamespaceIfNotExistsenableTableIfDisabled(String namespace, String name) throws IOException;

  /**
   * DeleteDisable the specified namespacetable if it exists.
   *is enabled.
   *
   * @param namespace the namespace of the table to disable
   * @param name the namespacename of the table to deletedisable
   * @throws IOException if a remote or network exception occurs
   */
@throws IllegalStateException ifvoid there are tables in the namespacedisableTableIfEnabled(String namespace, String name) throws IOException;

  /*/*
   * voidModify deleteNamespaceIfExists(String name) throws IOException;

  /**the specified table. The table must be disabled.
   *
Create the specified table* if@param itnamespace doesthe notnamespace exist.of the table to *modify
   * @param descriptorname the descriptorname forof the table to createmodify
   * @param splitKeysdescriptor the initial split keysdescriptor for the table
   * @throws IOException if a remote or network exception occurs
   */ @throws IllegalStateException if the specified table is not disabled
   void*/
createTableIfNotExists(TableDescriptor descriptor, @Nullable byte[][] splitKeys)
   void modifyTable(String namespace, String name, TableDescriptor descriptor) throws IOException;

  /**
   * EnableTruncate the specified table. The iftable itmust isbe disabled.
   *
   * @param namespace the namespace of the table to enabletruncate
   * @param name the name of the table to enabletruncate
   * @throws IOException if a remote or network exception occurs
   */ @throws  void enableTableIfDisabled(String namespace, String name) throws IOException;

  /**
   * Disable IllegalStateException if the specified table if ittable is enabled.not disabled
   */
  void * @param namespace the namespace of the table to disabletruncateTable(String namespace, String name) throws IOException;

  /**
   * Delete @paramthe nametable theif nameit ofexists. theThe table must tobe disabledisabled.
   *
@throws IOException if a* remote@param or networknamespace exceptionthe occursnamespace of the table */to delete
 void disableTableIfEnabled(String namespace,* String@param name) throwsthe IOException;table to   /**delete
   * Modify@throws theIOException specifiedif table.a Theremote tableor mustnetwork beexception disabled.occurs
   * @throws IllegalStateException if *the @paramspecified namespacetable theis namespacenot of thedisabled
table to modify */
  *void @param name thedeleteTableIfExists(String namespace, String name) of the table to modifythrows IOException;
 
  /**
   * Grant permissions @paramon descriptora thetable descriptorto forusers theor tablegroups.
   *
  @throws IOException* if@param anamespace remotethe ornamespace networkof exceptionthe occurstable
   * @throws@param IllegalStateExceptionname if the specifiedname tableof isthe not disabledtable
   */ @param permissions voidA modifyTable(String namespace, Stringmap from user name, TableDescriptorto descriptor)the throwspermissions IOException;for that user, given /**as a string containing
* Truncate the specified* table. The table must be disabled.    *    * @param namespace the namespace of the table to truncateonly the characters  * @param name the name of the table to truncate'a' (Admin), 'c' (Create), 'r' (Read), 'w' (Write), and 'x' (Execute).
   * @throws IOException if anything goes wrong
a remote or network*/
exception occurs void grantPermissions(String namespace, *String @throwsname, IllegalStateExceptionMap<String, ifString> thepermissions) specifiedthrows table is not disabled
   */
  void truncateTable(String namespace, String name) throws IOException;

  /**
   * Delete the table if it exists. The table must be disabled.
   *
   * @param namespace the namespace of the table to delete
   * @param name the table to delete
   * @throws IOException if a remote or network exception occurs
   * @throws IllegalStateException if the specified table is not disabled
   */
  void deleteTableIfExists(String namespace, String name) throws IOException;
}

To deploy your extension:

Create an extension directory

Code Block
$ mkdir -p /opt/cdap/master/ext/hbase/repl

Copy your jar to the directory

Code Block
$ cp myextension.jar /opt/cdap/master/ext/hbase/repl/

Modify cdap-site.xml to use your implementation of HBaseDDLExecutor

Code Block
<property> <name>hbase.ddlexecutor.extension.dir</name> <value>/opt/cdap/master/ext/hbase</value> </property>

Modify cdap-site.xml with any properties required by your executor. Anything prefixed by 'cdap.hbase.spi.hbase.' will be available through the Context object passed into your executor's initialize method

Code Block

<property>
  <name>cdap.hbase.spi.hbase.zookeeper.quorum</name>
  <value>hbase-master-i18003-1000.dev.continuuity.net:2181/cdap</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.zookeeper.session.timeout</name>
  <value>60000</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.cluster.distributed</name>
  <value>true</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.bulkload.staging.dir</name>
  <value>/tmp/hbase-staging</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.replication</name>
  <value>true</value>
</property>

...

IOException;
}

Sample implementation of the SPI is located here.

To deploy your extension, run these steps on both your master and slave clusters:

Create an extension directory:

Code Block
$ mkdir -p /opt/cdap/master/ext/hbase/repl

Copy your JAR to the directory:

Code Block
$ cp myextension.jar /opt/cdap/master/ext/hbase/repl/

Modify cdap-site.xml to use your implementation of HBaseDDLExecutor:

Code Block
<property> <name>hbase.ddlexecutor.extension.dir</name> <value>/opt/cdap/master/ext/hbase</value> </property>

Modify cdap-site.xml with any properties required by your executor. Anything prefixed by 'cdap.hbase.spi.hbase.' will be available through the Context object passed into your executor's initialize method:

Code Block

<property>
  <name>cdap.hbase.spi.hbase.zookeeper.quorum</name>
  <value>hbase-master-i18003-1000.dev.continuuity.net:2181/cdap</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.zookeeper.session.timeout</name>
  <value>60000</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.cluster.distributed</name>
  <value>true</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.bulkload.staging.dir</name>
  <value>/tmp/hbase-staging</value>
</property>
<property>
  <name>cdap.hbase.spi.hbase.replication</name>
  <value>true</value>
</property>

Before starting CDAP on the master cluster, run a command on the slave cluster to load the HBase coprocessors required by CDAP onto the slave's HDFS:
Code Block
slave$ cdap setup coprocessors
Start CDAP on the master cluster:
Code Block
master$ cdap master start

Manual Failover Procedure

Stop all CDAP programs on the master cluster
Stop CDAP on the master cluster
Copy any HDFS files that have not yet been copied using either your distro's solution or distcp

Run the replication status tool to get cluster state:

Code Block
master$ cdap run co.cask.cdap.data.tools.ReplicationStatusTool -m -o /tmp/master_state

Copy the master state onto your slave cluster:
Code Block
master$ scp /tmp/master_state <slave>:/tmp/master_state

Verify replication has copied the required data onto the slave:

Code Block
slave$ cdap run co.cask.cdap.data.tools.ReplicationStatusTool -i /tmp/master_state ... Master and Slave Checksums match. HDFS Replication is complete. HBase Replication is complete.

Run Hive's metatool to update locations for Hive tables:

Code Block

slave$ hive --service metatool -updateLocation hdfs://[slave-namenode-host]:[slave-namenode-port] hdfs://[master-namenode-host]:[master-namenode-port] -tablePropKey avro.schema.url -serdePropKey avro.schema.url

Start CDAP on the slave:
Code Block
slave$ cdap master start

Versions Compared

Old Version 6

New Version Current

Key

Cluster Setup

HBase

HDFS

Hive

HDFS

Hive

Kafka

CDAP Setup

Kafka

CDAP Setup