Oracle Data Integrationによるビッグ・データの管理

概要：Oracle Data IntegrationによるHadoopの統合

目的

Oracle Data Integrationを使用すると、データの存在場所がOracle Database、Hadoop、サード・パーティのデータベース、アプリケーション、ファイルのいずれであっても、またこれらのソースの組合せであっても、すべてのデータを移動したり変換したりできます。このデモでは、その方法について説明します。"一度設計すればどこでも実行できる"という方法論を採用すると、データ変換および移動の論理的ルールに焦点を合わせることが可能になり、実装方法の選択肢は独立させ、ユースケースの変化や新しいテクノロジーの公開に合わせて進化させることができます。

http://www.oracle.com/technetwork/database/bigdata-appliance/oracle-bigdatalite-2104726.html

所要時間

約1時間

はじめに

このチュートリアルは次の項に分かれています:

シナリオの確認
SqoopおよびOracle GoldenGateによるデータの取込み
Hive、Spark、またはPigによるデータ変換
Oracle Loader for HadoopによるOracle DBへのデータ・ロード
Big Data SQLによるOracleからのHadoopデータ・アクセス
Pigによるデータのセッション化
ODIパッケージによる全ステップの実行

シナリオ

Oracle MoviePlexはオンラインの映画ストリーミング会社です。この会社のWebサイトでは、顧客のあらゆる対話をJSON形式の大規模なログ・ファイルに収集しています。また、映画のデータをOracleソース・データベースで管理しています。これらのソースに含まれる情報を解放し、会社のデータウェアハウスに含まれるエンタープライズ・データと結合することで、顧客の行動、提供しているサービスの効果、Webサイトのコンテンツ編成などについての理解を深めることができるに違いありません。

Oracle Data Integrationには、すべてのデータの統合を宣言的に定義するための統一されたツール主導のアプローチが用意されています。"一度設計すればどこでも実行できる"という考え方は、実装言語とは無関係に統合プロセスを定義でき、さまざまな環境でプロセスを実行できることを意味します。現在RDBMS上で実行されている変換は、Hadoopクラスタの実行に再利用したり、将来、Hadoop言語や最適化が利用できるようになったときにそれらに利用したりできます。

Oracle MoviePlexで使用されている各種メカニズムを紹介します。ソース・データベースからHiveの表にデータがロードされます。このとき使用される方法は、SqoopとOracle Data Integrator（ODI）によるバルク・ロードと、Oracle GoldenGate（OGG）を使用したチェンジ・データ・キャプチャによるロードの両方です。ODIでは、Hive、Spark、またはPigを使用して結合、フィルタ、集計を適用してデータが変換されます。結果データは、最適化されたOracle Loader for Hadoop（OLH）またはOracle SQL Connector for Hadoop（OSCH）を使用してターゲットのOracle DBにアンロードできます。Oracle Big Data SQLを使用すると、HadoopデータもOracle DBで使用できます。ここではODIが、Hive表を公開するのに必要な外部表を、SQL問合せで使用されるOracle DBに透過的に生成します。

デモのリセット

/home/oracle/movie/moviework/odi/reset_ogg_odi.sh

準備ステップ

Big Data Lite仮想マシンのダウンロードとインストール

Big Data Lite 4.2のダウンロード、インストール、起動について詳しくは、Big Data Liteのデプロイメント・ガイド（英語）に書かれている手順を参照してください。

必要なサービスの起動

Oracle GoldenGate向けのOracle Databaseの準備

タスクバーにある「Terminal」アイコンを1回クリックしてターミナル・ウィンドウを開きます。Terminalウィンドウに次のコマンドを入力してセットアップ・スクリプトを実行します:
cd /home/oracle/movie/moviework/ogg
./enable_ogg_on_orcl.sh.

これを1回だけ実行しておけば、VMのリブート後も構成は引き続き有効です。このスクリプトで実行された変更を元に戻すには、同じディレクトリにあるdisable_ogg_on_orcl.shを実行します

Part 1 - SqoopおよびOracle GoldenGateによるデータの取込み

この項では、Sqoopのバルク・ロードとOracle GoldenGateのチェンジ・データ・キャプチャを使用して外部ソースからHadoopにデータを取り込む方法を説明します。

Oracle Data IntegratorでSqoopを使用したバルク・データの取込み

Oracle Data Integratorを使用すると、データの移動および変換に必要なマッピングを作成できます。この手順では、Oracle DBソースからHiveのターゲット表にデータをロードするために必要な既存のマッピングを確認して実行します。

Sqoopマッピングを開いて確認

startOdiStudio — デスクトップのツールバーからODIを起動します。

左側の「**Designer**」ナビゲータを選択し、**Projects**アコーディオンを開いて、**Big Data Demo**→**Demo**→**Mappings**→**A - Load Movie (Sqoop)**マッピングにナビゲートします。マッピングをダブルクリックして開きます。

Sqoopマッピングがマッピング・エディタに表示されます。Oracleのソース・データ・ストア**MOVIE**がHiveのターゲット・データ・ストア**movie_updates**にマッピングされている様子を確認できます。データ・ストアとは、表やデータの類似したエンティティを意味するODIの用語です。これはマッピングの論理ビューで、物理実装とは無関係に変換ルールが定義されています。
なお、movie_updatesには、操作の種類を表すopフィールドとタイムスタンプを表すtsフィールドが追加されています。これらの列は、Oracle GoldenGateによって追加される更新を使用して、後から実行されるリコンシリエーションのデータを準備するために使用されます。フィールドopは文字列'i'で初期化され、フィールドtsはOracleソース上で実行されたときにSYSDATEをコールします。

マッピングの下にある「**Physical**」タブをクリックして物理設計を確認します。

物理設計には、ソースとターゲットの実行単位であるデータベースやHadoopクラスタなどが表示されます。2つの表は別々のシステムに含まれ、アクセス・ポイント**MOVIE_AP**によってロードが制御されることを確認できます。

アクセス・ポイント「**MOVIE_AP**」を選択してデータ・ロードの構成を確認します。

**Properties**ウィンドウに移動します。このウィンドウは通常、マッピング・エディタの下にあります。**Loading Knowledge Module**セクションまでスクロール・ダウンし、セクションを展開します。ロード・ナレッジ・モジュール（LKM）として**LKM SQL to Hive SQOOP**が選択されています。Sqoopを使用したHadoopへのロードはこれによって実行されます。ナレッジ・モジュールの下には、Sqoopによるロードを構成したり、並列度を設定するなどしてチューニングしたりするためのオプションのリストがあります。AdvancedタブとParalellismタブでその他のオプションを確認してください。簡略化するために、オプションはすべてデフォルトのままにしますが、実際のユースケースでは、これらのオプションを使用してSqoopによるロードをチューニングします。

Sqoopのマッピングの実行

マッピング・エディタの上のタスクバーにある「**Run**」ボタンをクリックします。変更を保存するかどうか訊ねられたら「**Yes**」をクリックします。

Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、**Session started**というダイアログが表示されますので、ここでも「OK」をクリックします。

注：環境によっては、実行に数分かかることもあります。生成されるコードのみ表示するには、チェック・ボックス「**Simulation**」を選択します。そうすると、生成されたセッションがダイアログ・ウィンドウに表示され、オペレータには実行内容が何も表示されません。

実行内容を確認するには、**Operator**ナビゲータに移動して、**All Executions**ノードを開いて現在の実行を確認します。
実行が終了していない場合は、タスクが進行中であることを示す**Run**アイコンが表示されます。青色の「**Refresh**」アイコンをクリックしてビューをリフレッシュすることもできますが、ビューは5秒ごとに自動的にリフレッシュされます。

ロードが完了すると、警告アイコンが表示されます。警告アイコンが表示されてもこの実行では問題がなく、ロードが成功したことに変わりはありません。実行ツリーを開けば、実行に含まれる個々のタスクを確認できます。

Designerナビゲータに移動し、Models下の「**HiveMovie.movie**」を右クリックします。メニューから「**View Data**」を選択し、ロードされた行を確認します。

Hiveの**movie**表のすべての行がロードされた**Data**エディタが表示されます。

Oracle GoldenGate for Big Dataを使用した変更データの取込み

Oracle GoldenGateを使用すると、完了したトランザクションをソース・データベースから取得して、変更内容をターゲット・システムにレプリケートすることができます。Oracle GoldenGateは、システムに負荷をかけずに優れたパフォーマンスでこうした変更を確実に取得および適用します。Oracle GoldenGate for Big Dataは、GoldenGateで取得した変更をさまざまなHadoopターゲット・テクノロジーにレプリケートするためのコンポーネントです。このチュートリアルでは、OracleのMOVIE表への挿入を、対応するHiveのmovie_updates表にレプリケートします。Oracle GoldenGateはGoldenGate for Big Dataを通じてこの機能を提供し、Hive、HDFS、HBase、Flume、およびKafka向けのアダプタを提供します。この例では、最初にHiveの例を使用し、その後、Kafkaへの配信メカニズムのデモも行います。

Oracle GoldenGateの起動とOracleからHiveへのレプリケーションのセットアップ

次のコマンドを実行して、Capture（EMOV）プロセスを追加および起動します:
obey dirprm/bigdata.oby

次のコマンドを入力してggsciクライアントを閉じます:
exit

次のコマンドを実行して、Replicat（RMOV）プロセスを追加および起動します:
obey dirprm/bigdata.oby

Start Oracle sqlplus — 次のコマンドを実行して、新しく追加したプロセスのステータスを確認します:
info all

Kafkaが起動していない場合はRKAFKAプロセスがSTOPPEDと表示されますが、ここでは無視して構いません。メニュー・バーから2つ目のターミナル・ウィンドウを起動して、次のコマンドを入力します:
sqlplus system/welcome1@orcl

次のコマンドを実行して、Oracle表MOVIEに新しい行を挿入します:
INSERT INTO "MOVIEDEMO"."MOVIE" (MOVIE_ID, TITLE, YEAR, BUDGET, GROSS, PLOT_SUMMARY) VALUES ('1', 'Sharknado', '2014', '500000', '20000000', 'Flying sharks attack city');
commit;

注：代わりに次のコマンドを実行することもできます:
@ /home/oracle/movie/moviework/ogg/oracle_insert_movie.sql;

次のコマンドを実行して、Oracle表MOVIEの既存の行を更新します:
UPDATE "MOVIEDEMO"."MOVIE" SET BUDGET=BUDGET+'10000' where MOVIE_ID='1';
commit;

注：代わりに次のコマンドを実行することもできます:
@ /home/oracle/movie/moviework/ogg/oracle_update_movie.sql;

ODI Studioのデータ・ビューアに戻ります。再度スクリーンをリフレッシュし、最終行に移動します。操作タイプが"U"でBUDGETフィールドが前の行より10000多い510000のエントリが、新たに追加されていることを確認します。

次のコマンドを実行して、Oracle表MOVIEの既存の行を削除します:
DELETE FROM "MOVIEDEMO"."MOVIE" WHERE MOVIE_ID='1';
commit;

注：代わりに次のコマンドを実行することもできます:
@ /home/oracle/movie/moviework/ogg/oracle_delete_movie.sql;

ODI Studioのデータ・ビューアに戻ります。再度スクリーンをリフレッシュし、最終行に移動します。movie id 1のエントリとして、操作タイプが"D"の新しいエントリがもう1つあることを確認します。これは、表からの削除を意味します。

Start Hive CLI — メニュー・バーからターミナル・ウィンドウをもう1つ起動して、次のコマンドを入力します:
hive
これにより、結果表の問合せができるHiveのCLIが開きます。

Reconciliated view — 次のコマンドを入力します:
show create table movie_view;

これにより、Hiveのmovie表のリコンシリエーション済みバージョンの問合せに使用できるビューが表示されます。なお、このチュートリアルでは、リコンシリエーション済みの表を作成するためにOracle Data Integratorも使用します。Reconciliation by Viewはリアルタイムのステータスを提供しますが、頻繁に問い合わせるため、より多くのリソースが消費されます。
このビューは、次のコマンドを入力して問い合わせることができます。
select * from movie_view;

OracleからKafkaへのデータ・レプリケーション

Oracle GoldenGate for Big Dataには、GoldenGateで取得した変更をKafkaにレプリケートできるコンポーネントが組み込まれています。また、CSV形式だけでなく、JSON、XML、Avroなどさまざまな形式のデータを変換する機能を提供しています。Kafkaハンドラの例でのGoldenGateのプロセスは次のとおりです:

Kafkaのセットアップ

Start/Stop Servicesウィンドウで、次のサービスが起動していることを確認します:

ORCL Oracle Database 12c

Zookeeper

Kafka

Kafkaサーバーの起動

KafkaはZookeeperを使用するため、Zookeeperサーバーが起動していない場合は先に起動する必要があります。Zookeeperインスタンスが実行中でない場合は、Kafkaにパッケージされている便利なスクリプトを使用してシングル・ノードのZookeeperインスタンスを簡単に起動できます。

>ps -eaf | grep -i kafkaを実行して、zookeeperとKafkaが実行中かどうかを確認します。

Kafkaとzookeeperが実行中でない場合は、手順1に従って起動します。

トピックの作成
"oggtopic"という名前のkafkaトピックを作成します :

**cd /usr/lib/kafka**

**>bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic oggtopic --from-beginning**

テスト・メッセージの送信

Kafkaには、ファイルまたは標準入力から入力した内容をメッセージとしてKafkaクラスタに送信するコマンドライン・クライアントが付属しています。デフォルトでは、各行が独立したメッセージとして送信されます。

プロデューサを実行し、いくつかのメッセージを別のコンソールに入力してサーバーに送信します。

> /usr/lib/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic oggtopic
Test message 1
This is another test message 2

上の各コマンドを別のターミナルで実行している場合は、プロデューサ・ターミナルに入力したメッセージがコンシューマ・ターミナルに表示されるのを確認できます。

すべてのコマンドライン・ツールには追加オプションがあり、引数を付けずにコマンドを実行すると詳しい使用方法が表示されます。

Kafkaハンドラの操作

Oracleからのレプリケーションのセットアップ:
メニュー・バーの「**Terminal**」アイコンを1回クリックしてターミナル・ウィンドウを起動します。

ターミナル・ウィンドウで次のコマンドを実行します:
cd /u01/ogg-bd
./ggsci

ディレクトリ/u01/ogg-bd/dirprmでbigdata.obyファイルを実行してOGG Managerを起動し、Kafka Replicatを追加してKafka Replicatを起動します。
**ggsci> obey dirprm/bigdata.oby**

ビッグ・データのreplicatプロセスが稼働しているかどうか確認します:

**ggsci> info all**

次のコマンドを入力してOracle Database Extractプロセスを起動します:
cd /u01/ogg
./ggsci
>ggsci obey dirprm/bigdata.oby

Oracleのextractプロセスが稼働しているかどうか確認します:
**ggsci> info all**

メニュー・バーからX-ウィンドウを起動して、次のコマンドを入力します。:
**>sqlplus system/welcome1@orcl**

次のコマンドを実行して、Oracle表MOVIEに新しい行を挿入します:
INSERT INTO "MOVIEDEMO"."MOVIE" (MOVIE_ID, TITLE, YEAR, BUDGET, GROSS, PLOT_SUMMARY) VALUES ('1', 'Sharknado', '2014', '500000', '20000000', 'Flying sharks attack city');
commit;

*Kafkaのreplicatプロセスが生成したAVRO形式のKafkaメッセージは、Kafkaのコンシューマ・コンソールによって取得されます。*

Oracle Data IntegratorでHiveを使用して更新をマージ

このデモでは、SqoopとOracle GoldenGateによる初期データのロードと変更のリアルタイム・レプリケーションにOracle Data Integratorを使用しました。また、更新のリコンシリエーションを臨機応変に実行するためのビューも使用しました。ODIマッピングを使用してデータを一括でマージし、元のOracle表に保持されているのと同じデータを提供することもできます。

Hiveのマージ・マッピングを開いて確認

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。
左側の「Designer」ナビゲータを選択し、Projectsアコーディオンを開いて、Big Data Demo→Demo→Mappings→B - Merge Movies (Hive)マッピングにナビゲートします。マッピングをダブルクリックして開きます。

Merge Mapping — マッピングに使用される変換がマッピングの論理ビューに表示されます:

ソース・テーブル**movie_updates**には各映画に関する情報があり、opには操作タイプとしてI、U、およびDが、tsにはタイムスタンプ値が保持されています

**AGGREGATE**は、**movie_id**に基づいてすべての映画をグループ化する操作と最新のタイムスタンプを**max_ts**として計算する操作で使用されます

**JOIN**は、すべての映画更新エントリと、AGGREGATEで計算された各**movie_id**の最新のタイムスタンプを取得する操作で使用されます

**FILTER**は、操作タイプがDの行を除外する処理に使用されます。これは、最後の更新が削除だったことを意味します

なお、このマッピングでは、前の項で示したHiveビューと同じロジックを使用しています。このデモでは、リコンシリエーション・ビューと同様に、変換をグラフィカルに設計できるODIの機能について説明します。

Merge Physical — マッピングの物理ビューに切り替えます。変換はHive内で実行されるため、ソース・データ・ストアもターゲット・データ・ストアも同じ実行単位に含まれています。

Hiveマージ・マッピングの実行

マッピング・エディタの上のタスクバーにある「Run」ボタンをクリックします。変更を保存するかどうか訊ねられたら「Yes」をクリックします。

Merge Operator — Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、**Session started**というダイアログが表示されますので、ここでも「OK」をクリックします。

*注：環境によっては、実行に数分かかることもあります。生成されるコードのみ表示するには、チェック・ボックス「****Simulation****」を選択します。そうすると、生成されたセッションがダイアログ・ウィンドウに表示され、オペレータには実行内容が何も表示されません。*

実行内容を確認するには、**Operator**ナビゲータに移動して、**All Executions**ノードを開いて現在の実行を確認します。実行が終了していない場合は、タスクが進行中であることを示す**Run**アイコンが表示されます。青色の「**Refresh**」アイコンをクリックしてビューをリフレッシュすることもできますが、ビューは5秒ごとに自動的にリフレッシュされます。

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。タスク「**40 - Load MOVIE**」をダブルクリックすれば、生成されたコードをタスク・エディタで確認できます。

Designerナビゲータに移動してModelsを開き、「HiveMovie.movie」を右クリックします。メニューから「View Data」を選択し、ロードされた行を確認します。

Merge Data — Hiveの**movie**表のすべての行がロードされた**Data**エディタが表示されます。

Part 2 - Hive、Spark、およびPigを使用したデータ変換

この項では、SqoopとGoldenGateを使用して、事前にロードしておいた映画データをOracle Data Integratorを使用して変換します。このユースケースでは、表movieを顧客アクティビティ・イベント・データと結合します。このデータは、Flumeを使用して事前にHDFSファイルにロードされ、現在はHive表movieapp_log_odistageとして公開されています。アクティビティ・データにはレーティング・アクションが含まれているので、1つ1つの映画の平均レーティングを計算し、結果をHive表movie_ratingに格納します。論理マッピングを1つ作成し、ステージング場所の実行エンジンとしてHive、Spark、Pigのいずれを使用するか選択できます。ODIはHive SQL、Spark-Python、またはPig Latingのいずれかを生成し、適切なサーバー・エンジンでそれを実行します。

また、この章の一部として、ネストしたJSON HDFSファイルを入力として使用し、コンテンツ上で映画レーティングを計算できるようにフラット化するマッピングも示します。使用する実装エンジンはSparkです。

Hiveによる映画データの変換

Hiveのマッピングを開いて確認

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。

左側の「**Designer**」ナビゲータを選択し、**Projects**アコーディオンを開いて、**Big Data Demo→Demo→Mappings→C - Calc Ratings (Hive - Pig - Spark)**マッピングにナビゲートします。マッピングをダブルクリックして開きます。

マッピングに使用される変換がマッピングの論理ビューに表示されます:

ソース・テーブル**movie**には各映画に関する情報があり、ソース表**movieapp_log_avro**には未加工の顧客アクティビティが含まれています

**FILTER**は、**activity = 1**のイベント（レーティング・イベント）を抽出するのに使用されます

**AGGREGATE**は、すべてのレーティングを**movieid**に基づいてグループ化して、映画の平均レーティングを計算するのに使用されます

**JOIN**は、movie表にある映画の基本情報と集計済みのイベントを結合して、ターゲット表**movie_rating**に書き込むのに使用されます

ターゲット表**movie_rating**には、結合の結果が格納されます。これは、すべての言語で丸め処理を可能にするユーザー定義関数**XROUND**を使用します

マッピングの物理ビューに切り替え、下部のタブ**Hive**が選択されていることを確認します。3つの実装エンジンそれぞれに物理設計があります。変換はHive内で実行されるため、ソース・データ・ストアもターゲット・データ・ストアも同じ実行単位に含まれています。

ターゲット「**MOVIE_RATING**」を選択し、**Properties**ウィンドウのIntegration Knowledge Moduleタブを確認します。デフォルト設定で**IKM Hive Append**が選択されています。

Hiveマッピングの実行

Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、Session startedというダイアログが表示されますので、ここでも「OK」をクリックします。

注：環境によっては、実行に数分かかることもあります。生成されるコードのみ表示するには、チェック・ボックス「Simulation」を選択します。そうすると、生成されたセッションがダイアログ・ウィンドウに表示され、オペレータには実行内容が何も表示されません。

実行内容を確認するには、Operatorナビゲータに移動して、All Executionsノードを開いて現在の実行を確認します。実行が終了していない場合は、タスクが進行中であることを示すRunアイコンが表示されます。青色の「Refresh」アイコンをクリックしてビューをリフレッシュすることもできますが、ビューは5秒ごとに自動的にリフレッシュされます。

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。タスク「**40 - Load MOVIE_RATING**」をダブルクリックすれば、生成されたコードをタスク・エディタで確認できます。

Designerナビゲータに移動してModelsを開き、「HiveMovie.movie_rating」を右クリックします。
メニューから「View Data」を選択し、ロードされた行を確認します。

Hiveの**movie_rating**表のすべての行がロードされた**Data**エディタが表示されます。

Sparkによる映画データの変換

マッピングを開いて確認

ODI Studioを開き、Big Data Demo→Demo→Mappings→C - Calc Ratings (Hive - Pig - Spark)の順にナビゲートしてマッピングを開きます。詳しくは、前の章「Hiveによる映画データの変換」を参照してください。

なお、同じ論理マッピングをHive、Spark、Pigに使用します。これらの実装エンジンに対応した個別の論理設計はありません。論理ダイアグラムのプロパティ**Staging Location Hint**を使用して実装エンジンを選択してから、物理設計を生成します。

マッピングの物理ビューに切り替え、下部のタブ**Spark**が選択されていることを確認します。3つの実装エンジンそれぞれに物理設計があります。変換ではSparkを使用し、ソースとターゲットはHive Catalogメタデータを使用して定義されているため、ソース、ステージング（Spark）、ターゲット・データ・ストアは別々の実行単位で表示されます。

コンポーネント「**JOIN**」を選択し、**Properties**ウィンドウの**Extract Options**タブを確認します。Advancedタブに切り替えると、Extract Knowledge ModuleとしてXKM Spark Joinが自動的に選択されていることを確認できます。このようなXKMは、ユーザーに代わって各コンポーネントのSpark-Pythonコード生成を体系化します。Sparkのメモリの扱いや並列度をチューニングするには、**Options**タブでCACHE_DATA、NUMBER_OF_TASKSなどの詳細オプションを設定します。

Sparkマッピングの実行

Runダイアログで、Physical Mapping Designドロップダウン・メニューをHiveからSparkに変更します。Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、**Session started**というダイアログが表示されますので、ここでも「OK」をクリックします。

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。タスク「**30 - Load JOIN_AP**」をダブルクリックすれば、生成されたSpark-Pythonコードをタスク・エディタで確認できます。

結果データはデータ・ストアmovie_ratingで確認できます。詳しくは、前の章「Hiveによる映画データの変換」を参照してください。

Pigによる映画データの変換

マッピングを開いて確認

なお、同じ論理マッピングをHive、Spark、Pigに使用します。これらの実装エンジンに対応した個別の論理設計はありません。
論理ダイアグラムのプロパティ**Staging Location Hint**を使用して、次に生成する物理設計に対応する実装エンジンを選択します。

マッピングの物理ビューに切り替え、下部のタブ**Pig**が選択されていることを確認します。3つの実装エンジンそれぞれに物理設計があります。変換ではPigを使用し、ソースとターゲットはHive Catalogメタデータを使用して定義されているため、ソース、ステージング（Pig）、ターゲット・データ・ストアは別々の実行単位で表示されます。

Pigマッピングの実行

Pigマッピングを実行する前に、ターゲット表を切り捨てる必要があります。**Projects**ツリーの**Procedures**ノードに移動し、「**Truncate movie_rating**」を右クリックして「**Run...**」を選択してプロシージャを実行します。Runダイアログではデフォルト設定を使用し、以降のダイアログではすべて「OK」を選択します。

マッピング・エディタに戻り、マッピング・エディタの上のタスクバーにある「**Run**」ボタンをクリックします。

Runダイアログで、Physical Mapping Designドロップダウン・メニューをHiveからPigに変更します。Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、**Session started**というダイアログが表示されますので、ここでも「OK」をクリックします。

結果データはデータ・ストアmovie_ratingで確認できます。詳しくは、前の章「Hiveによる映画データの変換」を参照してください。

SparkによるネストしたJSONデータの変換

ソースとなるHDFS JSONデータの準備と確認

以降の変換に備えて、まずテスト用のJSONデータをHDFSに配置する必要があります。
ターミナル・ウィンドウで次のコマンドを実行します:

hdfs dfs -put ~/movie/moviework/odi/movie_ratings/movie_ratings.json /user/odi/demo/movie_ratings.json

注：このコピー手順をパッケージで自動化する方法については、パート6を参照してください

ターミナル・ウィンドウで次のコマンドを実行します:
hdfs dfs -cat /user/odi/demo/movie_ratings.json

映画とネストしたレーティング・レコード（1～5）を含むレコードが書き込まれたJSON形式のファイルが表示されます。これがマッピングの入力になります。データ・ストア**movie_ratings**（JSONファイルのメタデータを含む）を使用するODI Model **HDFSMovie**が表示されます。なお、**ratings**サブ配列のような複雑な属性に関するメタデータはデータ・ストアに保持されず、フラットなString型の属性として表示されます。

マッピングを開いて確認

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。

左側の「**Designer**」ナビゲータを選択し、**Projects**アコーディオンを開いて、Big Data Demo→Demo→Mappings→D - Calc Ratings（JSON Flatten）マッピングにナビゲートします。マッピングをダブルクリックして開きます。

マッピングに使用される変換がマッピングの論理ビューに表示されます:

ソースHDFSファイル**movie_ratings**には、各映画のレーティング・サブ配列とともに各映画の情報が、ネストしたJSON形式で収められています

Flattenは、レーティング情報のネストを解除するのに使用されます。各映画にネストされたレーティングのクロス積を計算します

AGGREGATEは、すべてのレーティングを**movie_id**に基づいてグループ化して、映画の平均レーティングを計算するのに使用されます

ターゲット表**movie_rating**には、結合の結果が格納されます。これは、すべての言語で丸め処理を可能にするユーザー定義関数XROUNDを使用します

マッピングの実行

Hiveの**movie_rating**表のすべての行がロードされた**Data**エディタが表示されます。movie_idが10、11、12、13、14のエントリを探してください。これらは、このマッピングによって追加されたものです。

Part 3 - Oracle Loader for HadoopによるOracle DBへのデータ・ロード

このタスクでは、先ほどのHive変換で得られた結果をHive表からOracle DBのデータウェアハウスにロードします。ロード処理には、Oracle DBに特に最適化されたメカニズムを使用する、Oracle Loader for Hadoopという組込みデータ・ローダーを使用します。

ODIマッピングとOracle Loader for HadoopによるOracle DBへの映画データのロード

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。

左側の「**Designer**」ナビゲータを選択し、**Projects**アコーディオンを開いて、Big Data Demo→Demo→Mappings→E - Load Oracle (OLH)マッピングにナビゲートします。マッピングをダブルクリックして開きます。

マッピングの論理ビューに表示されるのは、Hive表**movie**からOracle表MOVIEへのダイレクト・マップです。属性上の変換は実行されません。

マッピングの物理ビューに切り替えます。ここには、HiveソースからOracleターゲットへの移動が表示されます。

アクセス・ポイント「MOVIE_RATING_AP」**（ラベルの一部が隠れている場合があります）**を選択し、**Properties**ウィンドウのIntegration Knowledge Moduleタブを確認します。LKM Hive to Oracle OLH-OSCH Directが選択され、マッピングの反復実行を許可するTRUNCATE=Trueと、ステージング・ステップが追加されないようにするUSE_HIVE_STAGING_TABLE=False以外はデフォルト設定になっています。出力モードOLH_OUTPUT_MODEはデフォルトでJDBCに設定されています。これは、デバッグ用や単純なユースケースに適した設定です。その他の設定は、OCIまたはData Pumpを介したOLHロードを実行するため、またはOSCHを介したロードを実行するためのものです。

OLHマッピングの実行

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。さまざまなタスクをダブルクリックすれば、Loader ConfigとMapファイルがどのように生成されてローダーが実行されたのかを確認できます。

Designerナビゲータに移動してModelsを開き、「OracleMovie.ODI_MOVIE_RATING」を右クリックします。
メニューから「View Data」を選択し、ロードされた行を確認します。

OLH data — OracleのODI_MOVIE_RATING表のすべての行がロードされた**Data**エディタが表示されます。

Part 4 - Big Data SQLによるOracleからのHadoopデータ・アクセス

次の項では、Oracle Big Data SQLを使用する1つのマッピングで、Oracle DBとHiveの両方のデータをODIで変換します。Big Data SQLを使用すると、Oracleの機能豊富なSQL言語を使用してOracle ExadataからOracle Big Data Appliance上のデータをシームレスに問い合わせることができます。Oracle Databaseに含まれる他のすべてのデータとまったく同じ方法で、Hadoopに格納されているデータに対して問合せが実行されます。

このユースケースでは、先ほど使用したアクティビティ・データとOracle DBに格納されているCustomerデータを結合します。各顧客のすべての購買アクティビティを集計し、核となる顧客データにそれを結合します。

Big Data SQLを使用してHiveとOracle表から売上を計算

Big Data SQLマッピングを開いて確認

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。

左側の「Designer」ナビゲータを選択し、**Projects**アコーディオンを開いて、Big Data Demo→Demo→Mappings→F - Calc Sales（Big Data SQL）マッピングにナビゲートします。マッピングをダブルクリックして開きます。

マッピングに使用される変換がマッピングの論理ビューに表示されます:

Hiveソース・テーブル**movieapp_log_odistage**には各映画に関する情報があり、Oracleソース表**CUSTOMER**には各顧客の情報が含まれています

**FILTER**は、**activity = 11**のイベント（販売イベント）を抽出するのに使用されます

**JOIN**は、アクティビティ情報をそれぞれの顧客情報に結合するのに使用されます

**AGGREGATE**は、すべてのレーティングを**COUNTRY**と**CONTINENT**情報に基づいてグループ化し、所定の地域の全アクティビティの**sales**の合計値を計算するのに使用されます

マッピングの物理ビューに切り替えます。表**movieapp_log_stage**はHiveに含まれ、表**CUSTOMER**とすべての変換はターゲットOracle DBに含まれているのが分かります。

アクセス・ポイント「**MOVIEAPP**」を選択し、**Properties**ウィンドウのLoading Knowledge Moduleタブを確認します。外部表定義を介してOracleからHive表にリモート・アクセスできるように、**LKM Hive to Oracle (Big Data SQL)**が選択されています。LKMのオプションはすべてデフォルトです。

*注：LKM Hive to Oracle (Big Data SQL)はカスタムのサンプルKMで、Java.netからダウンロードできます*

カスタムLKMのコードは、Designer→Global Objects→Global Knowledge Modules→Loading→LKM Hive to Oracle (Big Data SQL)で確認できます。LKMの主要部分は、Big Data SQL構文を使用して外部表を生成するためのテンプレートです。この外部表は、行をターゲット表に挿入する統合問合せで使用されます。

Big Data SQLマッピングの実行

Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、Session startedというダイアログが表示されますので、ここでも「OK」をクリックします。

注：環境によっては、実行に数分かかることもあります。生成されるコードのみ表示するには、チェック・ボックス「Simulation」を選択します。そうすると、生成されたセッションがダイアログ・ウィンドウに表示され、オペレータには実行内容が何も表示されません。

実行内容を確認するには、Operatorナビゲータに移動して、All Executionsノードを開いて現在の実行を確認します。実行が終了していない場合は、タスクが進行中であることを示すRunアイコンが表示されます。青色の「Refresh」アイコンをクリックしてビューをリフレッシュすることもできますが、ビューは5秒ごとに自動的にリフレッシュされます。

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。

タスク「**40 - Create external table**」をダブルクリックして、生成されたBig Data SQLのコードをタスク・エディタで確認します。**Code**タブに切り替えます。タイプがORACLE_HIVEの外部表を作成する**create table**式を確認できます。この式は、Hiveのメタデータを使用してODIによって生成されたものです。

Designerナビゲータに移動してModelsを開き、「OracleMovie.ODI_COUNTRY_SALES」を右クリックします。
メニューから「View Data」を選択し、ロードされた行を確認します。

OracleのCOUNTRY_SALES表のすべての行がロードされた**Data**エディタが表示されます

Part 5 - Pigによるデータのセッション化

このデモでは、ODIとユーザー定義ファンクションとテーブル・ファンクションを使用できる機能を併用して、複雑なPigの複雑なマッピングを実行する方法を説明します。このマッピングでは、このHadoopディストリビューションに含まれるApache DataFuライブラリからPigのファンクションsessionizeを使用します。

このマッピングの目的は、Hive表movieapp_log_odistageに含まれるアクティビティをユーザー別に独立したセッションに整理し、地域別にセッションの継続時間の最小値、最大値、平均値の統計を算出することです。

ODIでのPigによるセッション化のロードとユーザー・アクティビティの分析

マッピングを開いて確認

ODI Studioを開き、**Big Data Demo→Demo→Mappings→G - Sessionize Data (Pig)**マッピングにナビゲートします。

マッピングに使用される変換がマッピングの論理ビューに表示されます:

Hiveソース表**movieapp_log_odistage**には各映画に関する情報があり、Hive表**cust**には各顧客の情報が含まれています

**SESSIONIZE**は、DataFu.sessionize UDFをコールするカスタムPigコードを含むテーブル・ファンクション・コンポーネントです。このファンクションは、アクティビティの発生時刻を含む時間枠に基づいてアクティビティをセッションに結合します

**AGGREGATE**は、場所（州、国）別にセッション統計を集計する処理に使用されます

**EXPRESSION**は、セッション統計を正規化および変換する処理に使用されます

**SORT**は、セッションの平均継続時間に基づいて集計レコードを順序付ける処理に使用されます

結果はターゲットのHive表**session_stats**に格納されます

マッピングの物理ビューに切り替えます。変換ではPigを使用し、ソースとターゲットはHive Catalogメタデータを使用して定義されているため、ソース、ステージング（Pig）、ターゲット・データ・ストアは別々の実行単位で表示されます。

物理コンポーネント「**SESSIONIZE**」を選択してPropertiesウィンドウに移動します。「**Extract Options**」タブを選択し、**PIG_SCRIPT_CONTENT**オプションの**Value**列をダブルクリックします。

このテーブル・ファンクションの実装に使用されるカスタムPigコードが、ダイアログに表示されます。この実装の核となるのは、DataFuのファンクション**Sessionize**です。なお、このコードでは、入力リレーションとして**MOVIEAPP_LOG_ODISTAGE**を使用し、出力リレーションとして**SESSIONIZE**を使用しています。

Pigマッピングの実行

Pigマッピングを実行する前に、ターゲット表を切り捨てることが必要な場合があります。ProjectsツリーのProceduresノードに移動し、「Truncate session_stats」を右クリックして「Run...」を選択してプロシージャを実行します。Runダイアログではデフォルト設定を使用し、以降のダイアログではすべて「OK」を選択します。

*注：この手順をパッケージで自動化する方法については、パート6を参照してください*。

Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、Session startedというダイアログが表示されますので、ここでも「OK」をクリックします。

注：環境によっては、実行に数分かかることもあります。生成されるコードのみ表示するには、チェック・ボックス「Simulation」を選択します。そうすると、生成されたセッションがダイアログ・ウィンドウに表示され、オペレータには実行内容が何も表示されません。

ロードが完了すると、セッションのすべてのタスクが成功したものとしてオペレータに表示されます。タスク「30 - Load JOIN_AP」をダブルクリックすれば、生成されたPigコードをタスク・エディタで確認できます。

結果データはデータ・ストアsession_statsで確認できます。

Part 6. ODIパッケージによる全ステップの実行

この項では、ODIパッケージを使用して、他のすべてのODIステップをまとめて順番に実行します。パッケージは、マッピング、プロシージャ、ツール、他のパッケージなどのさまざまなジョブを連携させるのに使用でき、条件付きロジックや変数を実行の一部としてパッケージに含めることもできます。このパッケージでは、JSONファイルをローカル・ファイル・システムからHDFSにコピーするFile Copyツールと、表の切捨てを実行するHive SQLコールを発行するプロシージャも使用します。

パッケージの確認

ODI Studioを開きます。ODI Studioを起動してログインする方法については、パート1を参照してください。

左側の「**Designer**」ナビゲータを選択し、**Projects**アコーディオンを開いて、**Big Data Demo→Demo→Packages→Big Data Load**パッケージにナビゲートします。パッケージをダブルクリックして開きます。

pkg — パッケージを確認します。マッピング、プロシージャ、ツールのコールがすべて1つのワークフローで接続され、1回の操作で実行されるようになっています。

「**Copy To HDFS**」ステップを選択し、プロパティを確認します。パラメータ**Source Logical Schema**は空で、**Target Logical Schema**は**HadoopLocal**に設定されています。こうすると、ソースはローカル・ファイル・システム・パスとして、ターゲットはHDFSパスとして定義されます。

「**C - Calc Ratings (Hive - Pig - Spark)**」ステップを選択してプロパティを確認します。Physical Mapping Designは**Hive**に設定されていますが、他のオプション（PigまたはSpark）に変更して別のエンジンで実行することもできます。

パッケージ・エディタの上のタスクバーにある「Run」ボタンをクリックします。変更を保存するかどうか訊ねられたら「Yes」をクリックします。

Runダイアログの「OK」をクリックします。すべてのデフォルトを使用して、ODI Studio UIに埋め込まれているローカル・エージェント上でこのマッピングを実行します。しばらくすると、Session startedというダイアログが表示されますので、ここでも「OK」をクリックします。

注：環境によっては、実行に数分かかることもあります

まとめ

このチュートリアルで学習した内容は、次のとおりです:

SqoopおよびOracle GoldenGateによるデータの取込み
Hive、Pig、およびSparkによるデータ変換
Oracle Loader for HadoopによるOracle DBへのデータ・ロード
Big Data SQLによるOracleからのHadoopデータ・アクセス
Pigによるデータのセッション化
ODIパッケージによる複数ジョブの連携

参考資料

Oracle Data Integration、Oracle Big Data SQL、Oracle Data WarehousingおよびOracle Analytical SQLについて詳しくは、Oracle Technology Networkを参照してください。

クレジット

著者: Alex Kotopoulis

To help navigate this Oracle by Example, note the following:

Hiding Header Buttons:: Click the Title to hide the buttons in the header. To show the buttons again, simply click the Title again.
Topic List Button:: A list of all the topics. Click one of the topics to navigate to that section.
Expand/Collapse All Topics:: To show/hide all the detail for all the sections. By default, all topics are collapsed
Show/Hide All Images:: To show/hide all the screenshots. By default, all images are displayed.
Print:: To print the content. The content currently displayed or hidden will be printed.