FluidQueryを使用してNetezza環境とHadoop環境のデータ連携やってみた。

###**FluidQueryとは何か？**
***
FluidQueryはIBM社のDWHアプライアンスであるPuredata System for Analytics(Netezza)のアドオン機能(無償)であり、Netezza環境からSQLを通して他のRDBMSやHadoop環境のデータを取得することができる非常に便利な機能です。興味深い点として、RDBMSの表データをNetezza側にコピーできることはもちろん、RDBMSに対して実行したSQL結果セットをNetezzaに持ってきたり、RDBMS側で実行したSQLの結果セットとNetezza側の表を結合させてユーザーが求める結果を取得するといったことも可能です。RDBMSの違いを意識することなく、1つの論理的なデータウェアハウスのような感じで様々なデータソースへのアクセスができるようになります。

また履歴データやアクセス負荷の低いものは他のHadoop環境にデータを移動させることが可能です。またHadoop上にあるデータに対してもSQLを解してNetezzaからデータを引っ張ってくることができます。
###**FluidQuery 1.6概要**
***
![FluidQuery1.6.png](https://qiita-image-store.s3.amazonaws.com/0/94920/37ba94e8-24a2-4917-4f58-88d02ce71eff.png)

![FluidQuery1.6_5.png](https://qiita-image-store.s3.amazonaws.com/0/94920/e4541e7b-a087-ee1c-1471-e5140760d448.png)

###**FluidQuery1.6を使用してNetezza環境とHadoop環境とのデータ連携をやってみた**
***
**テスト実施にあたっての事前準備作業**
・Netezza環境の準備:Netezza Emulatorのダウンロードとセットアップ
　参考:[PC上でメモリ5GBで動く! Netezza Software Emulatorを使ってみた]
        (http://http://qiita.com/ishida330/items/8a23ae32b34b9aa461a0)
・Hadoop環境の準備:IBM Biginsight 4.1のダウンロードとセットアップ
　参考:[評価版Hadoop環境:IBM BigInsight 4.1 はこちら参照 ](https://www.ibm.com/support/knowledgecenter/SSPT3X_4.1.0/com.ibm.swg.im.infosphere.biginsights.product.doc/doc/bi_qse.html)
・BigInsight4.1 とは何か？は下記資料参照ください。
　　参考:[BigInsight4.1 マニュアルサイト](http://www.ibm.com/support/knowledgecenter/SSPT3X_4.1.0/com.ibm.swg.im.infosphere.biginsights.welcome.doc/doc/welcome.html?lang=ja)
  　参考:[Introducing BigInsights, IBM's Hadoop- and Spark-based analytical   platform](http://www.slideshare.net/CynthiaSaracco/introducing-ibms-info)

今回は自前のクライアントPC環境にNetezza Emulatorを導入して、仮想Netezza環境をクライアントPC環境にセットアップしました。そしてHadoop環境としてIBM BigInsight 4.1評価版もダウンロードして同じくPC上に導入。クライアントPC上でNetezza のDBとBigInsightを稼動させ、データ連携ができるか試してみました。ちなみにFluidQueryを使用するためには、Netezza環境にINZAの導入が必要となります。今回はNPS 7.2.1なので、INZA 3.2.1をインストール済みの環境で実行しています。

![BigInsight_PDA_検証環境.png](https://qiita-image-store.s3.amazonaws.com/0/94920/a3ec233d-e53a-9fec-dec2-94b7b2426f12.png)
※FluidQueryのインストール前提としてのNPSバージョン、INZAパッケージのバージョンがあります。FluidQuery、INZAパッケージについてはNetezza代理店・サポートに問い合わせることで、確認、無償での入手可能です。

###**FluidQuery1.6のインストール**
***
入手したFluidQuery導入パッケージファイル(nz-fluidquery-v1.6.0.1.tar.gz)をHadoop環境とNetezza SMPホストの任意のディレクトリに保存、解凍作業をおこないます。

####**Hadoop環境へのインストール**

入手したFluidQuery導入パッケージファイル(nz-fluidquery-v1.6.0.0.tar.gz)を任意のディレクトリに保存、解凍作業をおこないます。

例:$ tar xzvf nz-fluidquery-version.tar.gz

次に解凍されたインストールファイルディレクトリ以下にある下記コマンドを実行します。

```lang:
[root@rvm IFQ1.6.0.1]# ./fluidquery_install.sh  
-------------------------------------------------------------------------------
IBM Fluid Query Installer
(C) Copyright IBM Corp. 2015  All rights reserved.
Version 1.6.0.1 [Build 160107-13]
-------------------------------------------------------------------------------
Installing data movement...
Data movement path: /fluidqueryLocal
Checking for previous installation of IBM Fluid Query data movement...
Previous installation of data movement does not exist in /fluidqueryLocal...
Creating HDFS directory /fluidquery...
HDFS directory /fluidquery created.
Copying files to /fluidqueryLocal...
Copying license to /fluidqueryLocal...
Copying uninstaller to /fluidqueryLocal...
Copying jars to HDFS...
BigInsight exists. Copying jars to BigSQL nodes...
install IFQ in BigSQL v3
 -> install IFQ on node: rvm.svl.ibm.com
 -> install IFQ on node: rvm.svl.ibm.com
IBM Fluid Query data movement installed succesfully in /fluidqueryLocal!
For details see the log file at /fluidqueryLocal/var/log/fluidquery_install/fluidquery_install.2016-02-20.07:28:06.log
[root@rvm IFQ1.6.0.1]#
```
※Hadoop側のFluidQueryは /fluidqueryLocal の下にファイルが配置されます。

####**Netezza環境へのインストール**
Netezza環境にも同様に同じファイルを展開して、解凍、インストール作業をおこいます。(下記はFluidQuery1.6から1.6.0.1へアップグレードしたため、少し出力は異なります。)

```lang:
[nz@netezza IFQ1.6.0.1]$ ./fluidquery_install.sh
-------------------------------------------------------------------------------
IBM Fluid Query Installer
(C) Copyright IBM Corp. 2015  All rights reserved.
Version 1.6.0.1 [Build 160107-13]
-------------------------------------------------------------------------------
Installing connector...
Checking if IBM Netezza Analytics is installed...
IBM Netezza Analytics is already installed. Ok.
Checking if IBM Netezza Analytics cartridge is installed...
IBM Netezza Analytics cartridge is already installed. Ok.
Checking for previous installation of IBM Fluid Query...
A previous installation of IBM Fluid Query was found.
Existing FluidQuery Version: 1.6.0.0 [Build 151110-258]
New FluidQuery Version: 1.6.0.1 [Build 160107-13]
Would you like to upgrade the existing IBM Fluid Query installation? (y/n) [y]
Backing up existing installation files of IBM Fluid Query to /nz/export/ae/products/fluidquery.backup.2016-04-06.12:49:56
Backup of the existing installation of IBM Fluid Query created...
Copying connector files...
Copying license to /opt/nz/licenses/ifq...
Copying uninstaller to /nz/export/ae/products/fluidquery...
Installation successful! Package installed in /nz/export/ae/products/fluidquery
For details see the log file at /nz/var/log/fluidquery_install/fluidquery_install.2016-04-06.12:49:52.log
[nz@netezza IFQ1.6.0.1]$
```
※Netezza側のFluidQueryは /nz/export/ae/products/fluidquery の下にファイルが配置されます。
以上でFluidQueryのインストールは完了です。簡単です！

###**FluidQueryを使用してHadoop上のデータを検索するための準備作業**
***
####**JDBCドライバファイルのセットアップ**
FluidQueryを使用するためには、JDBCドライバが必要となります。そのため、BigInsight4.1用のJDBCドライバをダウンロードし、所定の場所に配置します。今回JDBCドライバは db2jcc4.jar,db2jcc.jarをダウンロードして、下記ディレクトリに配置。

[BigInsight 4.1 JDBCダウンロードサイト](http://www-01.ibm.com/support/docview.wss?uid=swg21363866)
Netezza側 JDBCドライバ配置先：/nz/export/ae/products/fluidquery/libs/ibm/bigsql

####**fqConfigureの実行**
FluidQuery実行のためのセットアップ作業として、Netezza側で下記コマンドを実行、構成ファイルを作成します。BigInsight(Hadoop側)のIPアドレス情報とポート番号が必要となります。

```lang:
[nz@netezza fluidquery]$ cd /nz/export/ae/products/fluidquery
[nz@netezza fluidquery]$ ./fqConfigure.sh --host 192.168.153.134 --port  32051 --username bigsql --config nz2bigsql --provider ibm --service BIGSQL
```
####**fqRegisterの実行**
次に作成した構成ファイルをFluidQueryを使用するNetezza側のDBに登録します。今回はdemodbというDBに先ほどのコマンドで作成したnz2bigsqlという構成ファイルをnz2bigsqlというudtf名で登録します。

```lang:
[nz@netezza fluidquery]$ ./fqRegister.sh --udtf nz2bigsql --db demodb --config nz2bigsql
```

####**Netezzaからhadoop側のデータをクエリで取得する**
準備が整ったので、Netezza側のDBにログインして、BigInsight(Hadoop側)のデータをクエリで取得してみます。今回BigInsight側には付属のGOSALESDWスキーマのデモデータが入っているので、そちらのデータをクエリで取得してみます。

```lang:
DEMODB.ADMIN(ADMIN)=>  select * from TABLE WITH FINAL ( nz2bigsql ('','','SELECT COUNTRY_KEY,REGION_JA,COUNTRY_JA FROM GOSALESDW.go_region_dim'));
 COUNTRY_KEY |  REGION_JA   |   COUNTRY_JA
-------------+--------------+----------------
       90001 | アメリカ     | アメリカ
       90002 | アメリカ     | カナダ
       90003 | アメリカ     | メキシコ
       90004 | アメリカ     | ブラジル
       90005 | アジア太平洋 | 日本
       90006 | アジア太平洋 | シンガポール
       90007 | アジア太平洋 | 韓国
       90008 | アジア太平洋 | 中国
       90009 | アジア太平洋 | オーストラリア
       90010 | 北欧         | オランダ
       90011 | 北欧         | スエーデン
       90012 | 北欧         | デンマーク
       90013 | 北欧         | フィンランド
       90014 | 中欧         | フランス
       90015 | 中欧         | ドイツ
       90016 | 中欧         | スイス
       90017 | 中欧         | 英国
       90018 | 中欧         | ベルギー
       90019 | 南欧         | オーストリア
       90020 | 南欧         | イタリア
       90021 | 南欧         | スペイン
```
無事ちゃんと取得できましたね。これでNetezzaからHadoop上のデータに対して問い合わせをおこないっデータが取れることが確認できました。

###**Netezza環境のデータをHadoopに移動させる。**
***
FluidQueryはもう１つ面白い機能として、NetezzaからHadoopへ、HadoopからNetezzaへデータを圧縮した状態でパラレルデータ転送できるData Movement機能があるので、こちらも問題なく動くか試してみました。今回はHadoop環境からデータを移動させる手順を実施したので紹介してみます。

####**fq-import-conf.xml 構成ファイルの変更**

fq-import-confの編集を実施。HDFSとして事前に下記作業を実施しておきます。(BigInsight側作業) 今回はdemodbの全ての表を新規にHadoop側のHDFS(nz_table_dir ディレクトリ以下に作成)にコピーを作成するようにします。Data Movement機能は「Create」以外にも「Append」や「Incremental」のデータ移動が可能なので、増加したデータだけをHadoop側に退避することも可能となっています。またWhere句で条件指定をして、特定のレコードだけをHadoop側に退避させることも可能なようです。

```lang:
-- BigInsight側：/fluidqueryLocal/fq-import-conf.xml を編集する
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
        <!-- Execution configuration -->
    <property>
        <name>nz.fq.command</name>
        <value>import</value>
        <description>Type of data movement to transfer: import (NPS->Hadoop) or export (Hadoop->NPS)</description>
    </property>
    <!-- HDFS information -->
    <property>
        <name>nz.fq.output.path</name>
        <value>/user/bigsql/nz_table_dir</value>
        <description>Directory in HDFS where transferred data is stored</description>
    </property>
    <property>
        <name>nz.fq.format.fielddelim</name>
        <value>124</value>
        <description>Integer value of the single character field delimiter in the plain text output file</description>
    </property>
    <property>
        <name>nz.fq.format.null</name>
        <value>NULL</value>
        <description>Null value representation. Maximum 4 characters</description>
    </property>
    <property>
        <name>nz.fq.fs.temp</name>
        <value>/tmp</value
        <description>Location of a temporary files (logs, staus files etc) in HDFS</description>
    </property>
    <property>
        <name>nz.fq.hive.schema</name>
        <value>bigsql</value>
        <description>target schema name, under which all imported tables will be created. If id doesn't exist it will be automatically created</description>
    </property>
    <!-- Compression options -->
    <property>
        <name>nz.fq.compress</name>
        <value>true</value>
        <description>Transfer Netezza data in compressed internal format</description>
    </property>
    <property>
        <name>nz.fq.output.compressed</name>
        <value>true</value>
        <description>Indicates whether the data stored in Hadoop after transfer is in compressed internal format; depends on nz.fq.compress setting</description>
    </property>
    <!-- Netezza information -->
    <property>
        <name>nz.fq.nps.db</name>
        <value>demodb</value>
        <description>The Netezza database name; include double quotations around delimited database names</description>
    </property>
    <property>
        <name>nz.fq.tables</name>
        <value>*</value>
        <description>Comma-separated list of Netezza tables; include double quotations around delimited table names</description>
    </property>
    <property>
        <name>nz.fq.nps.server</name>
        <value>192.168.153.133</value>
        <description>The wall IP address or fully qualified hostname of the Netezza server</description>
    </property>
    <property>
        <name>nz.fq.nps.port</name>
        <value>5480</value>
        <description>The port number for the Netezza database instance NZ_DBMS_PORT</description>
    </property>
    <property>
        <name>nz.fq.nps.user</name>
        <value>admin</value>
        <description>The Netezza database user account name for access to the database</description>
    </property>
    <property>
        <name>nz.fq.nps.password</name>
        <value>password</value>
        <description>The password for the Netezza database user account</description>
    </property>
    <property>
        <name>nz.fq.nps.ssl</name>
        <value>false</value>
        <description>Netezza server connection type - when true then onlySecured JDBC mode is used; default is false</description>
    </property>
    <property>
        <name>nz.fq.nps.ssl.cacertificate</name>
        <value></value>
        <description>Full path to CA Certificate file stored on HDFS.
        CA Certificate file used to authenticate connections; used only when the SSL flag is true; if not provided then all connections will be accepted</description>
    </property>
    <property>
        <name>nz.fq.nps.where</name>
        <value></value>
        <description>The SQL WHERE clause to use for selecting the data to transfer</description>
    </property>
    <property>
        <name>nz.fq.splits</name>
        <value>12</value>
        <description>Number of concurrent JDBC load sessions to the Netezza host</description>
    </property>
    <property>
        <name>nz.fq.break.on.error</name>
        <value>false</value>
        <description>If it is set to true - stops import execution in case of error during the import of more than one table or whole database.</description>
    </property>
    <property>
        <name>nz.fq.append.mode</name>
        <value>create</value>
        <description>Defines handling of existing output data. Four modes: create, append, incremental, overwrite.
        create - new table is created, if old data exists import fails;
        append - data is appended to existing data, all records are added;
        incremental - only newer data from PDA table is imported;
        overwrite - if old data exists, it is deleted</description>
    </property>
    <property>
        <name>nz.fq.ignore.deleted</name>
        <value>false</value>
        <description>Valid only with nz.fq.append.mode set to incremental
        If false - stops incremental import execution, if any records on PDA table have been deleted or updated.</description>
    </property>
    <property>
        <name>nz.fq.sql.metadata</name>
        <value>true</value>
        <description>If true - creates table in Hive</description>
    </property>
    <!-- If missing HADOOP SQL properties. We will use local HIVE Runner
    Make sure to have hive-site.xml on HADOOP_CLASSPATH -->
</configuration>
```

####**Hadoop環境変数の設定**
Data Movement機能で使用するhadoop関連のJarファイルや構成ファイルへのパスを通すために環境変数を設定しておきます。今回はBigInsightの管理者ユーザーであるbigsqlユーザーに設定しておきます。(FluidQuery関連ファイルについても、権限をbigsqlユーザーに追加)

```lang:
$ export HADOOP_CLASSPATH="/usr/iop/current/hive-client/lib/*:/usr/ibmpacks/bigsql/4.1/hive/conf/:/fluidqueryLocal/nzjdbc3.jar"
```

####**データ移動コマンドの実行(Netezza -> Hadoop)**
以上で準備完了ですので、NetezzaからHadoopへのデータ移動(Import)を実行してみたいと思います。コマンドは下記の通りです。

```lang:
$hadoop jar /fluidqueryLocal/nzcodec.jar -conf fq-import-conf.xml
```
####**Hadoop環境で移動させたデータの検索を実施してみる**
それでは無事データ移動が完了したので、データをHIVE経由で参照してみたいと思います。サンプルとしてshohin表を１行検索してみました。(Hadoop環境ではNetezza環境の表データは「ユーザー名_表名」で保存されています。)

```lang:
[bigsql@rvm fluidqueryLocal]$ hive
・・・
hive> show schemas;
OK
bigsql
default
gosalesdw
Time taken: 2.822 seconds, Fetched: 3 row(s)
hive> use bigsql;
OK
Time taken: 0.147 seconds
hive> show tables;
OK
admin_customer
admin_meisai
admin_shohin
admin_shop
Time taken: 0.099 seconds, Fetched: 4 row(s)
hive> select * from admin_shohin limit 1;
OK
2       **      クレンザー、漂白剤、その他      台所洗剤        家庭用品
Time taken: 0.131 seconds, Fetched: 1 row(s)
hive>
```
ちゃんと結果が返ってきました。これだけでデータ移動がすむのは便利ですね。Netezza側で圧縮した状態でデータ転送され、Hadoop環境でも圧縮したデータに対してクエリを実行できるため、Netezza環境であまり参照されなくなった不要なデータをHadoop環境にアーカイブして保存したり、クエリの負荷をHadoop側にオフロードしたい場合に使えそうです。

###**Hadoop環境のデータをNetezzaに移動させる。　**
***
####**fq-export-conf.xml 構成ファイルの変更**
次は逆にHadoop環境のデータをNetezza側に移動させる機能を実行してみたいと思います。
こちらの設定もImportの時と同じくXML構成ファイルを編集します。今回はdemodb_no2というDBにadmin_customer表を戻します。admin_customer表は100万件の顧客データが入ったサンプル表となります。

```lang:
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
        <!-- Execution configuration -->
    <property>
        <name>nz.fq.command</name>
        <value>export</value>
        <description>Type of data movement to transfer: import (NPS->Hadoop) or export (Hadoop->NPS)</description>
    </property>

    <!-- HDFS information -->
    <property>
        <name>nz.fq.input.path</name>
        <value>/user/bigsql/nz_table_dir/admin_customer</value>
        <description>Directory in HDFS where retrieved data is stored</description>
    </property>
    <property>
        <name>nz.fq.format.fielddelim</name>
        <value>124</value>
        <description>Integer value of the single character field delimiter in the plain text output file</description>
    </property>
    <property>
        <name>nz.fq.format.null</name>
        <value>NULL</value>
        <description>Null value representation. Maximum 4 characters</description>
    </property>
    <property>
        <name>nz.fq.fs.temp</name>
        <value>/tmp</value>
        <description>Location of a temporary files (logs, staus files etc) in HDFS</description>
    </property>

    <!-- Netezza information -->
    <property>
        <name>nz.fq.nps.db</name>
        <value>demodb_no2</value>
        <description>The Netezza database name; include double quotations around delimited database names</description>
    </property>
    <property>
        <name>nz.fq.table</name>
        <value>ADMIN.customer</value>
        <description>The Netezza table; include double quotations around delimited table name</description>
    </property>
    <property>
        <name>nz.fq.append.mode</name>
        <value>append</value>
        <description>Defines handling of existing output data. Three modes: create, append, overwrite.
        create - new table is created, if old data exists import fails
        append - new data is appended to existing data
        overwrite - if old data exists, it is deleted</description>
    </property>
    <property>
        <name>nz.fq.nps.server</name>
        <value>192.168.153.133</value>
        <description>The wall IP address or fully qualified hostname of the Netezza server</description>
    </property>
    <property>
        <name>nz.fq.nps.port</name>
        <value>5480</value>
        <description>The port number for the Netezza database instance NZ_DBMS_PORT</description>
    </property>
    <property>
        <name>nz.fq.nps.user</name>
        <value>admin</value>
        <description>The Netezza database user account name for access to the database</description>
    </property>
    <property>
        <name>nz.fq.nps.password</name>
        <value>password</value>
        <description>The password for the Netezza database user account</description>
    </property>
    <property>
        <name>nz.fq.nps.ssl</name>
        <value>false</value>
        <description>Netezza server connection type - when true then onlySecured JDBC mode is used; default is false</description>
    </property>
    <property>
        <name>nz.fq.nps.ssl.cacertificate</name>
        <value></value>
        <description>Full path to CA Certificate file stored on HDFS.
        CA Certificate file used to authenticate connections; used only when the SSL flag is true; if not provided then all connections will be accepted</description>
    </property>
    <property>
        <name>nz.fq.exttab.columns</name>
        <value>*</value>
        <description>External table column names in proper order (for example: CODE,TITLE,PRICE), default is *</description>
    </property>
    <property>
        <name>nz.fq.exttab.schema</name>
        <value></value>
        <description>External table schema definition (for example: CODE CHAR(5), TITLE VARCHAR(255), PRICE INTEGER)
            default is empty string - the schema of the target table is used</description>
    </property>
    <property>
        <name>nz.fq.splits</name>
        <value>12</value>
        <description>Number of concurrent JDBC load sessions to the Netezza host</description>
    </property>
</configuration>
```

####**Hadoop環境変数の設定**
Exportを実施するときも、前回同様にData Movement機能で使用するhadoop関連のJarファイルや構成ファイルへのパスを通すために環境変数を設定しておきます。今回はBigInsightの管理者ユーザーであるbigsqlユーザーに設定しておきます。(FluidQuery関連ファイルについても、権限をbigsqlユーザーに追加)

```lang:
$ export HADOOP_CLASSPATH="/usr/iop/current/hive-client/lib/*:/usr/ibmpacks/bigsql/4.1/hive/conf/:/fluidqueryLocal/nzjdbc3.jar"
```

####**データ移動コマンドの実行(Hadoop -> Netezza)**
以上で準備完了ですので、NetezzaからHadoopへのデータ移動(Export)を実行してみたいと思います。コマンドは下記の通りです。

```lang:
$hadoop jar /fluidqueryLocal/nzcodec.jar -conf fq-export-conf.xml
```

####**Hadoop環境からNetezza環境に移動させたデータに対して検索を行う。**
問題なくデータ移動が完了したので、データをNetezza上のdemodb_no2というDBに表が入ったか見てみたいと思います。

```lang:
[nz@netezza fluidquery]$ nzsql -db demodb_no2
\Welcome to nzsql, the IBM Netezza SQL interactive terminal.

Type:  \h for help with SQL commands
       \? for help on internal slash commands
       \g or terminate with semicolon to execute query
       \q to quit

DEMODB_NO2.ADMIN(ADMIN)=> \d
          List of relations
 Schema |    Name     | Type  | Owner
--------+-------------+-------+-------
 ADMIN  | CUSTOMER    | TABLE | ADMIN
(1 rows)

DEMODB_NO2.ADMIN(ADMIN)=> select count(*) from CUSTOMER;
  COUNT
---------
 1000000
(1 row)
```
表もちゃんと存在しています。結果件数もデータ移動前と同じ件数が返ってきましたので、これでデータ移動機能の基本動作は簡単だけど確認できたかな。今回はHadoop側からデータ移動を実行するジョブを実行していますが、Netezza側からもRDBMSのファンクションとして登録すれば実行できるインターフェイスも用意されているようです。Netezza<->Hadoop間のデータ移動機能はかなり様々な用途で活用できそうですし、今回確認できていない機能もマニュアルを読む限りあるので、それらはおいおい試していきたいと思います。