Um tutorial HDFS para analistas de dados presos em bancos de dados relacionais

Introdução

Até agora, você provavelmente já ouviu falar do Hadoop Distributed File System (HDFS), especialmente se você for analista de dados ou alguém responsável por mover dados de um sistema para outro. No entanto, quais são os benefícios que o HDFS tem sobre os bancos de dados relacionais?

O HDFS é uma solução escalável e de código aberto para armazenar e processar grandes volumes de dados. O HDFS provou ser confiável e eficiente em muitos data centers modernos.

O HDFS utiliza hardware comum junto com software de código aberto para reduzir o custo geral por byte de armazenamento.

Com sua replicação integrada e resiliência a falhas de disco, o HDFS é um sistema ideal para armazenar e processar dados para análise. Ele não requer os fundamentos e a sobrecarga para dar suporte à atomicidade, consistência, isolamento e durabilidade da transação (ACID), como é necessário com os sistemas de banco de dados relacionais tradicionais.

Além disso, quando comparado com bancos de dados corporativos e comerciais, como Oracle, a utilização do Hadoop como plataforma de análise evita custos extras de licenciamento.

Uma das perguntas que muitas pessoas fazem quando aprendem pela primeira vez sobre o HDFS é: Como faço para colocar meus dados existentes no HDFS?

Neste artigo, examinaremos como importar dados de um banco de dados PostgreSQL para o HDFS. Usaremos o Apache Sqoop, que atualmente é a solução de código aberto mais eficiente para transferir dados entre HDFS e sistemas de banco de dados relacionais. O Apache Sqoop foi projetado para carregar dados em massa de um banco de dados relacional para o HDFS (importação) e para gravar dados em massa do HDFS para um banco de dados relacional (exportação).

As etapas neste tutorial foram escritas para alguém com conhecimento básico de execução de consultas SQL e conhecimento básico de comandos HDFS.

O sistema de banco de dados utilizado é o PostgreSQL 9.5 para Windows, e a versão HDFS é Cloudera Hadoop 2.5.0-cdh5.2.0 em uma máquina virtual Centos 6.4 Linux.

O Apache Sqoop conta com os arquivos JAR do driver JDBC que são específicos do fornecedor do banco de dados relacional e da versão do banco de dados.

Para executar as etapas mostradas neste artigo, o usuário precisará de permissões para se conectar remotamente ao banco de dados PostgreSQL, permissões SELECT no banco de dados relacional, permissões de gravação no HDFS e permissões de execução no executável Sqoop.

Para o propósito deste tutorial, criamos um banco de dados PostgreSQL, chamado de Toptal , e o tornamos acessível através da porta 5432.

Fonte de dados PostgreSQL

Para começar, em nosso banco de dados PostgreSQL Toptal , criaremos uma tabela de dados de teste chamada sales . Vamos supor que o certificado OpenSSL e os arquivos de chave privada já existam no servidor PostgreSQL.

 Server [localhost]: Database [postgres]: Toptal Port [5432]: Username [postgres]: Password for user postgres: psql (9.5.3) Toptal=# create table sales Toptal-# ( Toptal(# pkSales integer constraint salesKey primary key, Toptal(# saleDate date, Toptal(# saleAmount money, Toptal(# orderID int not null, Toptal(# itemID int not null Toptal(# ); CREATE TABLE

Em seguida, vamos inserir 20 linhas na tabela:

 Toptal=# insert into sales values (1, '2016-09-27', 1.23, 1, 1); INSERT 0 1 Toptal=# insert into sales values (2, '2016-09-27', 2.34, 1, 2); INSERT 0 1 Toptal=# insert into sales values (3, '2016-09-27', 1.23, 2, 1); INSERT 0 1 Toptal=# insert into sales values (4, '2016-09-27', 2.34, 2, 2); INSERT 0 1 Toptal=# insert into sales values (5, '2016-09-27', 3.45, 2, 3); INSERT 0 1 Toptal=# insert into sales values (6, '2016-09-28', 3.45, 3, 3); INSERT 0 1 Toptal=# insert into sales values (7, '2016-09-28', 4.56, 3, 4); INSERT 0 1 Toptal=# insert into sales values (8, '2016-09-28', 5.67, 3, 5); INSERT 0 1 Toptal=# insert into sales values (9, '2016-09-28', 1.23, 4, 1); INSERT 0 1 Toptal=# insert into sales values (10, '2016-09-28', 1.23, 5, 1); INSERT 0 1 Toptal=# insert into sales values (11, '2016-09-28', 1.23, 6, 1); INSERT 0 1 Toptal=# insert into sales values (12, '2016-09-29', 1.23, 7, 1); INSERT 0 1 Toptal=# insert into sales values (13, '2016-09-29', 2.34, 7, 2); INSERT 0 1 Toptal=# insert into sales values (14, '2016-09-29', 3.45, 7, 3); INSERT 0 1 Toptal=# insert into sales values (15, '2016-09-29', 4.56, 7, 4); INSERT 0 1 Toptal=# insert into sales values (16, '2016-09-29', 5.67, 7, 5); INSERT 0 1 Toptal=# insert into sales values (17, '2016-09-29', 6.78, 7, 6); INSERT 0 1 Toptal=# insert into sales values (18, '2016-09-29', 7.89, 7, 7); INSERT 0 1 Toptal=# insert into sales values (19, '2016-09-29', 7.89, 8, 7); INSERT 0 1 Toptal=# insert into sales values (20, '2016-09-30', 1.23, 9, 1); INSERT 0 1

Vamos selecionar os dados para verificar se os dados parecem corretos:

 Toptal=# select * from sales; pksales | saledate | saleamount | orderid | itemid ---------+------------+------------+---------+-------- 1 | 2016-09-27 | $1.23 | 1 | 1 2 | 2016-09-27 | $2.34 | 1 | 2 3 | 2016-09-27 | $1.23 | 2 | 1 4 | 2016-09-27 | $2.34 | 2 | 2 5 | 2016-09-27 | $3.45 | 2 | 3 6 | 2016-09-28 | $3.45 | 3 | 3 7 | 2016-09-28 | $4.56 | 3 | 4 8 | 2016-09-28 | $5.67 | 3 | 5 9 | 2016-09-28 | $1.23 | 4 | 1 10 | 2016-09-28 | $1.23 | 5 | 1 11 | 2016-09-28 | $1.23 | 6 | 1 12 | 2016-09-29 | $1.23 | 7 | 1 13 | 2016-09-29 | $2.34 | 7 | 2 14 | 2016-09-29 | $3.45 | 7 | 3 15 | 2016-09-29 | $4.56 | 7 | 4 16 | 2016-09-29 | $5.67 | 7 | 5 17 | 2016-09-29 | $6.78 | 7 | 6 18 | 2016-09-29 | $7.89 | 7 | 7 19 | 2016-09-29 | $7.89 | 8 | 7 20 | 2016-09-30 | $1.23 | 9 | 1 (20 rows)

Os dados parecem bons, então vamos prosseguir.

Importar para o HDFS usando Sqoop

Com a fonte de dados definida, agora estamos prontos para importar os dados para o HDFS. O comando sqoop que examinaremos está listado abaixo e detalharemos cada argumento nos marcadores a seguir. Observe que o comando deve estar em uma linha completa ou, como mostrado abaixo, com a barra invertida (o caractere de continuação da linha de comando do Linux) no final de cada linha, exceto a última.

 sqoop import --connect 'jdbc:postgresql://aaa.bbb.ccc.ddd:5432/Toptal?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory' \ --username 'postgres' -P \ --table 'sales' \ --target-dir 'sales' \ --split-by 'pksales'

• sqoop import - O executável é denominado sqoop e estamos instruindo-o a importar os dados de uma tabela ou visualização de um banco de dados para o HDFS.
• --connect - Com o argumento --connect , estamos passando a string de conexão JDBC para o PostgreSQL. Nesse caso, usamos o endereço IP, o número da porta e o nome do banco de dados. Também precisamos especificar que o SSL está sendo utilizado e fornecer a classe SSLSocketFactory a ser usada.
• --username - Neste exemplo, o nome de usuário é um login do PostgreSQL, não um login do Windows. O usuário deve ter permissões para se conectar ao banco de dados especificado e selecionar na tabela especificada.
• -P - Isso solicitará ao usuário da linha de comando a senha. Se o Sqoop raramente for executado, essa pode ser uma boa opção. Existem várias outras maneiras de passar a senha para o comando automaticamente, mas estamos tentando simplificar para este artigo.
• --table - É aqui que passamos o nome da tabela PostgreSQL.
• --target-dir - Este argumento especifica o diretório HDFS onde os dados devem ser armazenados.
• --split-by - Devemos fornecer ao Sqoop um identificador exclusivo para ajudá-lo a distribuir a carga de trabalho. Mais tarde na saída do trabalho, veremos onde o Sqoop seleciona os valores mínimo e máximo para ajudar a definir os limites de divisão.

É uma boa ideia colocar o comando em um script para fins de repetibilidade e edição, conforme mostrado abaixo:

 [hdfs@localhost:/sqoop]$ cat sqoopCommand.sh sqoop import --connect 'jdbc:postgresql://aaa.bbb.ccc.ddd:5432/toptal?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory' \ --username 'postgres' -P \ --table 'sales' \ --target-dir 'sales' \ --split-by 'pksales' [hdfs@localhost:/sqoop]$

Agora, é hora de executar o script de comando Sqoop acima. A saída do comando Sqoop é mostrada abaixo.

 [hdfs@localhost:/sqoop]$ ./sqoopCommand.sh 16/10/02 18:58:34 INFO sqoop.Sqoop: Running Sqoop version: 1.4.5-cdh5.2.0 Enter password: 16/10/02 18:58:40 INFO manager.SqlManager: Using default fetchSize of 1000 16/10/02 18:58:40 INFO tool.CodeGenTool: Beginning code generation 16/10/02 18:58:41 INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM "sales" AS t LIMIT 1 16/10/02 18:58:41 INFO orm.CompilationManager: HADOOP_MAPRED_HOME is /usr/lib/hadoop-0.20-mapreduce Note: /tmp/sqoop-training/compile/77f9452788024792770d52da72ae871f/sales.java uses or overrides a deprecated API. Note: Recompile with -Xlint:deprecation for details. 16/10/02 18:58:43 INFO orm.CompilationManager: Writing jar file: /tmp/sqoop-training/compile/77f9452788024792770d52da72ae871f/sales.jar 16/10/02 18:58:43 WARN manager.PostgresqlManager: It looks like you are importing from postgresql. 16/10/02 18:58:43 WARN manager.PostgresqlManager: This transfer can be faster! Use the --direct 16/10/02 18:58:43 WARN manager.PostgresqlManager: option to exercise a postgresql-specific fast path. 16/10/02 18:58:43 INFO mapreduce.ImportJobBase: Beginning import of sales 16/10/02 18:58:45 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 16/10/02 18:58:46 INFO db.DBInputFormat: Using read commited transaction isolation 16/10/02 18:58:46 INFO db.DataDrivenDBInputFormat: BoundingValsQuery: SELECT MIN("pksales"), MAX("pksales") FROM "sales" 16/10/02 18:58:47 INFO mapred.JobClient: Running job: job_201609280401_0005 16/10/02 18:58:48 INFO mapred.JobClient: map 0% reduce 0% 16/10/02 18:59:04 INFO mapred.JobClient: map 50% reduce 0% 16/10/02 18:59:14 INFO mapred.JobClient: map 75% reduce 0% 16/10/02 18:59:15 INFO mapred.JobClient: map 100% reduce 0% 16/10/02 18:59:18 INFO mapred.JobClient: Job complete: job_201609280401_0005 16/10/02 18:59:18 INFO mapred.JobClient: Counters: 23 16/10/02 18:59:18 INFO mapred.JobClient: File System Counters 16/10/02 18:59:18 INFO mapred.JobClient: FILE: Number of bytes read=0 16/10/02 18:59:18 INFO mapred.JobClient: FILE: Number of bytes written=1190344 16/10/02 18:59:18 INFO mapred.JobClient: FILE: Number of read operations=0 16/10/02 18:59:18 INFO mapred.JobClient: FILE: Number of large read operations=0 16/10/02 18:59:18 INFO mapred.JobClient: FILE: Number of write operations=0 16/10/02 18:59:18 INFO mapred.JobClient: HDFS: Number of bytes read=438 16/10/02 18:59:18 INFO mapred.JobClient: HDFS: Number of bytes written=451 16/10/02 18:59:18 INFO mapred.JobClient: HDFS: Number of read operations=4 16/10/02 18:59:18 INFO mapred.JobClient: HDFS: Number of large read operations=0 16/10/02 18:59:18 INFO mapred.JobClient: HDFS: Number of write operations=4 16/10/02 18:59:18 INFO mapred.JobClient: Job Counters 16/10/02 18:59:18 INFO mapred.JobClient: Launched map tasks=4 16/10/02 18:59:18 INFO mapred.JobClient: Total time spent by all maps in occupied slots (ms)=48877 16/10/02 18:59:18 INFO mapred.JobClient: Total time spent by all reduces in occupied slots (ms)=0 16/10/02 18:59:18 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0 16/10/02 18:59:18 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0 16/10/02 18:59:18 INFO mapred.JobClient: Map-Reduce Framework 16/10/02 18:59:18 INFO mapred.JobClient: Map input records=20 16/10/02 18:59:18 INFO mapred.JobClient: Map output records=20 16/10/02 18:59:18 INFO mapred.JobClient: Input split bytes=438 16/10/02 18:59:18 INFO mapred.JobClient: Spilled Records=0 16/10/02 18:59:18 INFO mapred.JobClient: CPU time spent (ms)=3980 16/10/02 18:59:18 INFO mapred.JobClient: Physical memory (bytes) snapshot=481574912 16/10/02 18:59:18 INFO mapred.JobClient: Virtual memory (bytes) snapshot=2949685248 16/10/02 18:59:18 INFO mapred.JobClient: Total committed heap usage (bytes)=127401984 16/10/02 18:59:18 INFO mapreduce.ImportJobBase: Transferred 451 bytes in 33.7555 seconds (13.3608 bytes/sec) 16/10/02 18:59:18 INFO mapreduce.ImportJobBase: Retrieved 20 records. [hdfs@localhost:/sqoop]$

Observe que a última linha de saída acima mostra que 20 registros foram recuperados, o que corresponde aos 20 registros na tabela do banco de dados PostgreSQL.

Após executar o comando Sqoop, podemos executar o comando hdfs dfs -ls para ver o diretório que foi criado por padrão com o nome da tabela no HDFS.

 [hdfs@localhost:/sqoop]$ hdfs dfs -ls Found 1 items drwxrwxrwx - toptal data 0 2016-10-02 18:59 sales [hdfs@localhost:/sqoop]$

Podemos usar o comando hdfs dfs -ls novamente para listar o conteúdo do diretório de sales . Se você observar o HDFS, poderá notar que os dados são particionados e espalhados por quatro arquivos por padrão, não apenas contidos em um.

 [hdfs@localhost:/sqoop]$ hdfs dfs -ls sales Found 6 items -rw-rw-rw- 1 toptal data 0 2016-10-02 18:59 sales/_SUCCESS drwxrwxrwx - toptal data 0 2016-10-02 18:58 sales/_logs -rw-rw-rw- 1 toptal data 110 2016-10-02 18:59 sales/part-m-00000 -rw-rw-rw- 1 toptal data 111 2016-10-02 18:59 sales/part-m-00001 -rw-rw-rw- 1 toptal data 115 2016-10-02 18:59 sales/part-m-00002 -rw-rw-rw- 1 toptal data 115 2016-10-02 18:59 sales/part-m-00003 [hdfs@localhost:/sqoop]$

O comando hdfs dfs -cat exibirá todos os registros na primeira partição dos dados de vendas no HDFS.

 [hdfs@localhost:/sqoop]$ hdfs dfs -cat sales/part-m-00000 1,2016-09-27,1.23,1,1 2,2016-09-27,2.34,1,2 3,2016-09-27,1.23,2,1 4,2016-09-27,2.34,2,2 5,2016-09-27,3.45,2,3 [hdfs@localhost:/sqoop]$

Observe que o delimitador de arquivo padrão é uma vírgula. Além disso, observe que há apenas cinco linhas em cada partição, porque as 20 linhas na origem foram distribuídas igualmente pelas quatro partições.

Para limitar o número de linhas que são exibidas na tela, podemos canalizar a saída do comando cat para o comando head como mostrado abaixo, para verificar o conteúdo das outras três partições.

O argumento -n 5 para o comando head limita a saída da tela às primeiras cinco linhas.

(Observe que, no nosso caso, isso é desnecessário, pois há apenas cinco linhas em cada partição para começar. Na prática, porém, você provavelmente terá muito mais linhas do que isso em cada partição e desejará verificar apenas as primeiras para certifique-se de que eles pareçam corretos, então isso mostra como fazer isso.)

 [hdfs@localhost:/sqoop]$ hdfs dfs -cat sales/part-m-00001 |head -n 5 6,2016-09-28,3.45,3,3 7,2016-09-28,4.56,3,4 8,2016-09-28,5.67,3,5 9,2016-09-28,1.23,4,1 10,2016-09-28,1.23,5,1 [hdfs@localhost:/sqoop]$ hdfs dfs -cat sales/part-m-00002 |head -n 5 11,2016-09-28,1.23,6,1 12,2016-09-29,1.23,7,1 13,2016-09-29,2.34,7,2 14,2016-09-29,3.45,7,3 15,2016-09-29,4.56,7,4 [hdfs@localhost:/sqoop]$ hdfs dfs -cat sales/part-m-00003 |head -n 5 16,2016-09-29,5.67,7,5 17,2016-09-29,6.78,7,6 18,2016-09-29,7.89,7,7 19,2016-09-29,7.89,8,7 20,2016-09-30,1.23,9,1 [hdfs@localhost:/sqoop]$

Se você precisar executar uma consulta para extrair dados de várias tabelas no banco de dados PostgreSQL, isso pode ser feito com o seguinte comando:

 [hdfs@localhost:/sqoop]$ cat sqoopCommand.sh sqoop import --connect 'jdbc:postgresql://aaa.bbb.ccc.ddd:5432/toptal?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory' \ --username 'postgres' -P \ --target-dir 'creditCardOrders' \ --split-by 'pksales' \ --query "select s.pksales, s.saledate, s.saleamount, o.shippingtype, o.methodofpayment from sales s inner join orders o on s.orderid=o.orderid where o.methodofpayment='credit card' and \$CONDITIONS" [hdfs@localhost:/sqoop]$

No comando acima, usamos alguns dos mesmos argumentos para o comando Sqoop, mas eles assumem importância diferente quando usados ​​com um comando SQL.

• --target-dir - O diretório de destino informa ao Sqoop em qual diretório no HDFS armazenar os dados selecionados. Esse argumento é exigido pelo Sqoop ao usar uma consulta de formato livre.
• --split-by - Embora estejamos selecionando a chave primária da tabela de vendas, ainda temos que fornecer ao Sqoop um identificador exclusivo para ajudá-lo a distribuir a carga de trabalho.
• --query - Este é o argumento no qual fornecemos a consulta SQL. A consulta acima está entre aspas duplas. Observe que não há uma barra invertida (o caractere de continuação de linha) nas várias linhas que contêm a consulta. Observe também o and \$CONDITIONS no final da cláusula WHERE . Isso é exigido pelo Sqoop porque o Sqoop substituirá automaticamente o token $CONDITIONS por uma expressão exclusiva.

Problemas ou nenhum problema: você deve considerar o HDFS

O HDFS tem muitas vantagens sobre os bancos de dados relacionais. Se você estiver fazendo análise de dados, considere migrar seus dados para o HDFS hoje.

Com as habilidades aprendidas aqui, importar dados de um sistema de banco de dados relacional para o HDFS é um processo simples e direto que pode ser realizado com um único comando. Embora esses exemplos tenham um pequeno número de linhas, a mecânica de importação de grandes volumes de dados para HDFS de uma tabela de banco de dados PostgreSQL permanece a mesma.

Você pode até experimentar importar tabelas grandes e delimitadores de armazenamento variados. Usar o Apache Sqoop é mais eficiente do que exportar os dados do banco de dados para um arquivo, transferir o arquivo do servidor de banco de dados para o HDFS e, em seguida, carregar o arquivo para o HDFS.