リレーショナルデータベースで立ち往生しているデータアナリストのためのHDFSチュートリアル

公開: 2022-03-11

序章

これまでに、Hadoop分散ファイルシステム（HDFS）について聞いたことがあると思います。特に、データアナリストや、あるシステムから別のシステムへのデータの移動を担当する人の場合はなおさらです。 しかし、HDFSがリレーショナルデータベースよりも優れている点は何ですか？

HDFSは、大量のデータを保存および処理するためのスケーラブルなオープンソースソリューションです。 HDFSは、多くの最新のデータセンターで信頼性と効率性が高いことが証明されています。

HDFSは、コモディティハードウェアとオープンソースソフトウェアを利用して、ストレージのバイトあたりの全体的なコストを削減します。

組み込みのレプリケーションとディスク障害に対する回復力を備えたHDFSは、分析用のデータを保存および処理するための理想的なシステムです。 従来のリレーショナルデータベースシステムで必要とされるトランザクションのアトミック性、一貫性、分離、および耐久性（ACID）をサポートするための基盤とオーバーヘッドは必要ありません。

さらに、Oracleなどのエンタープライズおよび商用データベースと比較した場合、分析プラットフォームとしてHadoopを利用することで、追加のライセンスコストを回避できます。

多くの人がHDFSについて最初に学ぶときに尋ねる質問の1つは、既存のデータをHDFSに取り込むにはどうすればよいですか。

この記事では、PostgreSQLデータベースからHDFSにデータをインポートする方法を検討します。 現在最も効率的なオープンソースソリューションであるApacheSqoopを使用して、HDFSとリレーショナルデータベースシステム間でデータを転送します。 Apache Sqoopは、リレーショナルデータベースからHDFSにデータを一括ロード（インポート）し、HDFSからリレーショナルデータベースにデータを一括書き込み（エクスポート）するように設計されています。

HDFS

データをHDFSに移行することにより、分析を高速化します。
つぶやき

このチュートリアルの手順は、SQLクエリの実行に関する基本的な知識とHDFSコマンドの基本的な知識を持っている人を対象としています。

使用されるデータベースシステムはWindows用のPostgreSQL9.5であり、HDFSバージョンはCentos6.4Linux仮想マシン上のClouderaHadoop2.5.0-cdh5.2.0です。

Apache Sqoopは、リレーショナルデータベースベンダーとデータベースバージョンに固有のJDBCドライバーJARファイルに依存しています。

この記事に示されている手順を実行するには、ユーザーはPostgreSQLデータベースにリモート接続するためのアクセス許可、リレーショナルデータベースでのSELECTアクセス許可、HDFSでの書き込みアクセス許可、およびSqoop実行可能ファイルでの実行アクセス許可が必要です。

このチュートリアルでは、PostgreSQLデータベースを作成し、 Toptalという名前を付けて、ポート5432からアクセスできるようにしました。

PostgreSQLデータソース

まず、PostgreSQL Toptalデータベースで、 salesという名前のテストデータテーブルを作成します。 OpenSSL証明書と秘密鍵ファイルが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

次に、テーブルに20行を挿入します。

 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

データを選択して、データが正しく表示されることを確認しましょう。

 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)

データは良さそうなので、先に進みましょう。

Sqoopを使用してHDFSにインポートする

データソースが定義されたら、データをHDFSにインポートする準備が整いました。 検討するsqoopコマンドを以下に示します。以下の箇条書きで各引数を分類します。 コマンドは1行にあるか、以下に示すように、最後を除く各行の最後に円記号（Linuxコマンドラインの継続文字）が付いていることに注意してください。

 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実行可能ファイルの名前はsqoopで、テーブルまたはビューからデータベースからHDFSにデータをインポートするように指示しています。
  • --connect --connect引数を使用して、PostgreSQLのJDBC接続文字列を渡します。 この場合、IPアドレス、ポート番号、およびデータベース名を使用します。 また、SSLが使用されていることを指定し、使用するSSLSocketFactoryクラスを提供する必要があります。
  • --usernameこの例では、ユーザー名はPostgreSQLログインであり、Windowsログインではありません。 ユーザーには、指定されたデータベースに接続し、指定されたテーブルから選択するための権限が必要です。
  • -Pこれにより、コマンドラインユーザーにパスワードの入力を求めるプロンプトが表示されます。 Sqoopがめったに実行されない場合、これは良いオプションかもしれません。 コマンドにパスワードを自動的に渡す方法は他にも複数ありますが、この記事ではパスワードをシンプルに保つようにしています。
  • --tableこれは、PostgreSQLテーブルの名前を渡す場所です。
  • --target-dirこの引数は、データが保存されるHDFSディレクトリを指定します。
  • --split-byワークロードの分散を支援するために、Sqoopに一意の識別子を提供する必要があります。 ジョブ出力の後半で、Sqoopが分割境界の設定に役立つ最小値と最大値を選択する場所を確認します。

以下に示すように、再現性と編集の目的で、コマンドをスクリプトに入れることをお勧めします。

 [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]$

次に、上記のSqoopコマンドスクリプトを実行します。 Sqoopコマンドの出力を以下に示します。

 [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]$

上記の出力の最後の行は、20レコードが取得されたことを示しています。これは、PostgreSQLデータベースのテーブルにある20レコードに対応します。

Sqoopコマンドを実行した後、 hdfs dfs -lsコマンドを実行して、HDFSのテーブル名でデフォルトで作成されたディレクトリを確認できます。

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

hdfs dfs -lsコマンドを再度使用して、 salesディレクトリの内容を一覧表示できます。 HDFSを見ると、データが1つに含まれているだけでなく、デフォルトで4つのファイルに分割および分散されていることがわかります。

 [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]$

hdfs dfs -catコマンドは、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]$

デフォルトのファイル区切り文字はコンマであることに注意してください。 また、ソースの20行が4つのパーティションに均等に分散されているため、各パーティションには5行しかないことに注意してください。

画面に出力される行数を制限するために、以下に示すように、 catコマンドの出力をheadコマンドにパイプして、他の3つのパーティションの内容を確認できます。

headコマンドの-n 5引数は、画面出力を最初の5行に制限します。

（この場合、最初は各パーティションに5行しかないため、これは不要です。ただし、実際には、各パーティションにはこれよりもはるかに多くの行があり、最初の数行を確認するだけです。それらが正しく見えることを確認してください。そうすれば、その方法がわかります。）

 [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]$

PostgreSQLデータベース内の複数のテーブルからデータを抽出するためにクエリを実行する必要がある場合、これは次のコマンドで実行できます。

 [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]$

上記のコマンドでは、Sqoopコマンドと同じ引数のいくつかを使用していますが、SQLコマンドで使用すると重要性が異なります。

  • --target-dirターゲットディレクトリは、選択したデータを保存するHDFS上のディレクトリをSqoopに指示します。 この引数は、自由形式のクエリを使用するときにSqoopで必要になります。
  • --split-by -salesテーブルの主キーを選択していますが、ワークロードの分散を支援するために、Sqoopに一意の識別子を提供する必要があります。
  • --queryこれは、SQLクエリを提供する引数です。 上記のクエリは二重引用符で囲まれています。 クエリを含む複数の行には円記号（行継続文字）がないことに注意してください。 また、 WHERE句の最後にあるand \$CONDITIONSにも注意してください。 Sqoopは$CONDITIONSトークンを一意の式に自動的に置き換えるため、これはSqoopで必要です。

問題があるか問題がない：HDFSを検討する必要があります

HDFSには、リレーショナルデータベースに比べて多くの利点があります。 データ分析を行っている場合は、今日、データをHDFSに移行することを検討する必要があります。

ここで学んだスキルを使用すると、リレーショナルデータベースシステムからHDFSにデータをインポートすることは、単一のコマンドで実行できる単純で簡単なプロセスです。 これらの例には少数の行がありますが、PostgreSQLデータベーステーブルからHDFSに大量のデータをインポートするメカニズムは同じです。

大きなテーブルをインポートしたり、ストレージ区切り文字を変更したりすることもできます。 Apache Sqoopを使用すると、データベースデータをファイルにエクスポートし、ファイルをデータベースサーバーからHDFSに転送してから、ファイルをHDFSにロードするよりも効率的です。

関連： Rでデータの変更を後押し