Пост 2. Sqoop импорт из реляционной базы данных в Hadoop.

Доброго времени суток! В своем первом посте я высказал мою точку зрения относительно Hadoop и Hive, теперь предлагаю перейти от философии к более конкретным вещам. Hive,как правило, работает с структурированными данными (таблицами). Зачастую данные попадают в Hive из реляционной базы данных (что то вроде архива, обеспечение жизненного цикла хранилища). Каким образом я могу это сделать? Голь на выдумку хитра, и существует множество способов проделать эту операцию. Я Вам расскажу об инструменте, предлагаемом cloudera – sqoop. Немного теории, что же умеет импортировать sqoop и какие вещи необходимо о нем знать:

1) Атомарной единицей импорта может быть:

- Партиция таблицы (поддерживает условие where)

- Одна таблица RDBMS

- Вся RDBMS

2) Sqoop умеет делать все это со всеми JDBC совместимыми базами. 

3) Импортирует из базы в HDFS ввиде плоских файлов (CSV) либо во внутреннем формате Hadoop Sequence File.

4) Поддерживает инкрементальный импорт данных

Можно долго еще описывать теоретическую часть, но в контексте sqoop будет полезнее посмотреть практический пример. Давайте импортируем данные из Oracle RDBMS.

Процесс импорта будет выглядеть примерно вот так:

Из реляционной базы данные попадают в hadoop, a далее файлы перемещаются внутри hadoop в дирректорию hive

Схема стенда в моем случае была следующей:

 Hadoop Client - отдельно выделенная машина с экземпляром ПО hadoop, установленном на нем Cloudera manager, Hue, Hive, Sqoop. Никаких вычислительных задач и задач хранения эта нода не выполняет (ни DataNode, TackTracker процессов там не запущено). С этой машины в дальнейшем я буду работать с этой машины с кластером Hadoop (9 серверов) и сервером баз данных Oracle RDBMS.  Давайте импортируем таблицу из базы данных Oracle в Hadoop. Для примера я взял стандартную схему scott и стандартную таблицу emp.

1) Устанавливаем на машине «Hadoop client» sqoop.

[root@cdh ~]# yum install sqoop

2) Шаг второй, опциональный – проверка JDBC соединения c базой данных. 

Для этого нам понадобится написать маленькую Java программку

[root@cdh ~]# cat OracleJDBC.java

import java.sql.DriverManager;

import java.sql.Connection;

import java.sql.SQLException;

public class OracleJDBC {

        public static void main(String[] argv) {

                System.out.println("-------- Oracle JDBC Connection Testing ------");

                try {

                        Class.forName("oracle.jdbc.driver.OracleDriver");

                } catch (ClassNotFoundException e) {

                        System.out.println("Where is your Oracle JDBC Driver?");

                        e.printStackTrace();

                        return;

                }

                System.out.println("Oracle JDBC Driver Registered!");

                Connection connection = null;

                try {

                        connection = DriverManager.getConnection(

                                        "jdbc:oracle:thin:@db11203x64:1521:orcl", "scott",tiger");

                } catch (SQLException e) {

                        System.out.println("Connection Failed! Check output console");

                        e.printStackTrace();

                        return;

                }

                if (connection != null) {

                        System.out.println("You made it, take control your database now!");

                } else {

                        System.out.println("Failed to make connection!");

                }

        }

}

[root@cdh oracle]# javac OracleJDBC.java

[root@cdh oracle]# java -classpath /home/oracle/ojdbc6.jar:/home/oracle/ OracleJDBC

-------- Oracle JDBC Connection Testing ------

Oracle JDBC Driver Registered!

You made it, take control your database now!

[root@cdh oracle]#

Ура! Коннект есть!

3) На всех серверах кластера в /etc/hosts/ прописываем сервер базы данных (в моем случае это было echo >> “192.168.1.108 db11203x64”). 

Прим.: для подобных вещей удобно использовать утилиту pdsh

4) Импортируем из базы данные Oracle в Hadoop

[root@cdh oracle]# sqoop import --connect jdbc:oracle:thin:@//db11203x64:1521/orcl --username scott -password tiger --table emp  --target-dir /tmp/oracle_dump --columns "EMPNO,DEPTNO" -m 1

,где db11203x64 – имя сервера баз данных, 1521 – порт прослушки listener, orcl – SID Oracle RDBMS

5) Просмотрим результат

[root@cdh oracle]# hadoop fs -ls /tmp/oracle_dump

Found 3 items

-rw-r--r--   3 root hdfs          0 2012-11-02 09:05 /tmp/oracle_dump/_SUCCESS

drwxrwxrwt   - root hdfs          0 2012-11-02 09:04 /tmp/oracle_dump/_logs

-rw-r--r--   3 root hdfs        112 2012-11-02 09:04 /tmp/oracle_d

[root@cdh oracle]# hadoop fs -cat /tmp/oracle_dump/part-m-00000

12/11/02 10:22:08 INFO util.NativeCodeLoader: Loaded the native-hadoop library

7369,20

7499,30

7521,30

7566,20

7654,30

7698,30

6) Ну а теперь напрямую в hive
[root@cdh oracle]# sqoop import --connect jdbc:oracle:thin:@//db11203x64:1521/orcl --username scott -password tiger --table emp --columns "EMPNO,DEPTNO" --hive-import --warehouse-dir /user/beeswax/warehouse --fields-terminated-by ',' --split-by EMPNO --hive-table emp -m 1Импорт «напрямую» подразумевает добавление мета информации в базу данных Hive.

7)  Давайте импортируем еще одну таблицу - побольше. Сначала мы ее создадим.
На сервере базы данных:
[oracle@db11203x64 ~]$ sqlplus scott/tiger@orcl

SQL*Plus: Release 11.2.0.3.0 Production on Wed Nov 7 02:31:52 2012

Copyright (c) 1982, 2011, Oracle.  All rights reserved.

Connected to:

Oracle Database 11g Enterprise Edition Release 11.2.0.3.0 - 64bit Production

With the Partitioning, Automatic Storage Management, OLAP, Data Mining

and Real Application Testing options

SQL> create table test_for_dump(a number,b number,c varchar2(256));

SQL> begin

  for i in 1..10000000 loop

  insert into test_for_dump values(i,i*8,'ramdom string');

  end loop;

  commit;

  end;

/

SQL> select bytes/(1024*1024) from user_segments where segment_name='TEST_FOR_DUMP';

BYTES/(1024*1024)
-----------------
              341

Ну вот уже не плохо таблица 341 Мб.

На сервере клиента hadoop

[root@cdh oracle]# sqoop import --connect jdbc:oracle:thin:@//db11203x64:1521/orcl --username scott -password tiger --table test_for_dump --columns "A,B,C" --hive-import --warehouse-dir /user/beeswax/warehouse --fields-terminated-by ',' --split-by A --hive-table sqoop_test  -m 1

Время выполнения - 1min43sec. Как бы мне ускорить этот поцесс? Распараллелить! Меняем в настройках   -m 1 на   -m 9 это означает, что к базе данных будут подключаться не один мэпер (сервер), а 9 и в параллель тащить данные от туда.

[root@cdh oracle]# sqoop import --connect jdbc:oracle:thin:@//db11203x64:1521/orcl --username scott -password tiger --table test_for_dump --columns "A,B,C" --hive-import --warehouse-dir /user/beeswax/warehouse --fields-terminated-by ',' --split-by A --hive-table sqoop_test  -m 9

Время выполнения - 31sec, намного лучше. Как правило при импорте данных на большой кластер Hadoop узким местом становится дисковая подсистема баз данных.

8) Ну и напоследок давайте рассмотрим инкрементальную загрузку данных (догрузку)
на сервере баз данных:
SQL>  insert into test_for_dump values(10000001,1,'');

1 row created.

SQL> commit;

Commit complete.

На клиенте Hadoop
[12:28]hdfs@cdh:~$ sqoop import --connect jdbc:oracle:thin:@//db11203x64:1521/orcl --username scott -password tiger --table test_for_dump --columns "A,B,C" --hive-import --warehouse-dir /user/beeswax/warehouse --fields-terminated-by ',' --split-by A --hive-table sqoop_test -m 9 --check-column A --incremental append --last-value 10000000

Догрузилась одна строка. Ура!
До новых постов!
Если У вас есть вопросы - не стесняйтесь их задавать!