SparkでのHDFSコネクタの使用

はじめに

この記事では、Hadoop Distributed File System (HDFS)コネクタをSparkアプリケーション・フレームワークとともに使用する方法を説明するウォークスルーを提供します。ウォークスルーでは、Oracle Linux 7.4オペレーティング・システムを使用し、Sparkを1つのコンピュータでスタンドアロンとして実行します。

前提条件

ウォークスルーを完了するための前提条件は、次のとおりです:

コンピュート・インスタンスを作成する権限が必要です。ガイダンスは、「インスタンスの作成」を参照してください。
起動したサービス・インスタンスに接続できる必要があります。ガイダンスは、インスタンスへの接続を参照してください。
オブジェクト・ストレージとの対話に使用するIdentity and Access Management (IAM)ユーザーについて、適切なOCID、フィンガープリント、および秘密キーが必要です。ガイダンスは、「設定および前提条件」を参照してください。
接続できるオブジェクト・ストレージ・バケットが必要です。
IAMユーザーは、コンソールを使用してそのバケットに対する読取りと書込みが可能である必要があります。

Sparkの使用

Sparkおよび依存関係のインストール

ノート

この例のために、Sparkを現在のユーザーのホーム・ディレクトリにインストールします。本番シナリオでは、これは行わないようにしてください。

ノート

バージョン2.7.7.0以上では、必要なサードパーティのすべての依存関係はインストールされなくなりました。必要なサードパーティ依存関係は、zipアーカイブのthird-party/libフォルダにバンドルされているため、手動でインストールする必要があります。

Computeサービスのインスタンスを作成します。ガイダンスは、「インスタンスの作成」を参照してください。
Secure Shell (SSH)接続を使用して接続できるように、サービス・インスタンスにパブリックIPアドレスがあることを確認します。ガイダンスは、インスタンスへの接続を参照してください。
SSH接続を使用してサービス・インスタンスに接続します。
次のコード例を使用して、Sparkおよびその依存関係、JavaおよびScalaをインストールします。

# We'll use wget to download some of the artifacts that need to be installed
sudo yum install wget
 
# First install Java
sudo yum install java-1.8.0-openjdk.x86_64
export JAVA_HOME=/usr/lib/jvm/jre-1.8.0-openjdk
# Should be something like: OpenJDK Runtime Environment (build 1.8.0_161-b14)
java -version
 
# Then install Scala
wget https://downloads.lightbend.com/scala/2.12.4/scala-2.12.4.rpm
sudo yum install scala-2.12.4.rpm
# Should be something like: Scala code runner version 2.12.4 -- Copyright 2002-2017, LAMP/EPFL and Lightbend, Inc.
scala -version
 
# Then download Spark
wget https://archive.apache.org/dist/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz
tar xvf spark-2.2.1-bin-hadoop2.7.tgz
export SPARK_HOME=$HOME/spark-2.2.1-bin-hadoop2.7
export PATH=$PATH:$SPARK_HOME/bin
 
# Start a Spark master
cd $SPARK_HOME
./sbin/start-master.sh

HDFSコネクタのダウンロードと構成ファイルの作成

ノート

この例のために、JARとキー・ファイルを現在のユーザーのホーム・ディレクトリに配置します。本番シナリオでは、かわりに、適切な権限が設定された(つまり、SparkとHiveが実行されているユーザーが読み取ることができる)共通の場所にこれらのファイルを配置します。

次のコード例を使用して、HDFSコネクタをサービス・インスタンスにダウンロードし、関連する構成ファイルを追加します。その他の情報は、オブジェクト・ストレージでのHDFSコネクタを参照してください。

wget https://github.com/oracle/oci-hdfs-connector/releases/download/v2.9.2.1/oci-hdfs.zip
unzip oci-hdfs.zip -d oci-hdfs
 
cd $HOME
mkdir .oci
# Create or copy your API key into the $HOME/.oci directory
 
cd $SPARK_HOME/conf
# Create a core-site.xml (e.g. by transferring one you have, using vi etc.). Consult
# https://docs.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#Properties
# for what this should look like
 
# Create a spark-defaults.conf file from the template
cp spark-defaults.conf.template spark-defaults.conf

spark-defaults.confファイルで、最後に次の行を追加します:

spark.sql.hive.metastore.sharedPrefixes= shaded.oracle,com.oracle.bmc

データの準備

データをテストする場合は、MovieLensデータ・セットを使用します。

https://grouplens.org/datasets/movielens/latest/で最新のデータ・セットをダウンロードします。必ず小さいデータ・セット(Small)をダウンロードしてください。
ダウンロードしたファイルを解凍します。
movies.csvファイルをオブジェクト・ストレージ・バケットにアップロードします。

Sparkシェルを使用したテスト

データの準備ができた状態で、Sparkシェルを起動し、サンプル・コマンドを使用してテストできるようになりました:

cd $SPARK_HOME
./bin/spark-shell
 
scala> sc.wholeTextFiles("oci://PipedUploadTest@sampletenancy/")
java.io.IOException: No FileSystem for scheme: oci

この時点では、oci://ファイル・システム・スキーマが使用できないため、エラーが表示されます。Sparkシェルを起動する前に、JARファイルを参照する必要があります。これを行う例を次に示します:

./bin/spark-shell --jars $HOME/oci-hdfs/lib/oci-hdfs-full-1.2.7.jar --driver-class-path $HOME/oci-hdfs/lib/oci-hdfs-full-1.2.7.jar
 
scala> sc.wholeTextFiles("oci://PipedUploadTest@sampletenancy/")
res0: org.apache.spark.rdd.RDD[(String, String)] = oci://PipedUploadTest@sampletenancy/ MapPartitionsRDD[1] at wholeTextFiles at <console>:25
 
scala> sc.textFile("oci://PipedUploadTest@sampletenancy/movies.csv").take(20).foreach(println)
movieId,title,genres
1,Toy Story (1995),Adventure|Animation|Children|Comedy|Fantasy
2,Jumanji (1995),Adventure|Children|Fantasy
3,Grumpier Old Men (1995),Comedy|Romance
4,Waiting to Exhale (1995),Comedy|Drama|Romance
5,Father of the Bride Part II (1995),Comedy
6,Heat (1995),Action|Crime|Thriller
7,Sabrina (1995),Comedy|Romance
8,Tom and Huck (1995),Adventure|Children
9,Sudden Death (1995),Action
10,GoldenEye (1995),Action|Adventure|Thriller
11,"American President, The (1995)",Comedy|Drama|Romance
12,Dracula: Dead and Loving It (1995),Comedy|Horror
13,Balto (1995),Adventure|Animation|Children
14,Nixon (1995),Drama
15,Cutthroat Island (1995),Action|Adventure|Romance
16,Casino (1995),Crime|Drama
17,Sense and Sensibility (1995),Drama|Romance
18,Four Rooms (1995),Comedy
19,Ace Ventura: When Nature Calls (1995),Comedy

コマンドは正常に実行され、オブジェクト・ストレージに接続できます。コマンドの実行のたびに--jars引数を渡したくない場合は、かわりにoci-hdfs-full JARファイルを$SPARK_HOME/jarsディレクトリにコピーできることに注意してください。

Spark Thrift Serverの起動

ポート10015でSpark Thrift Serverを起動し、次に示すようにBeelineコマンドライン・ツールを使用して、JDBC接続を確立してから、基本的な問合せを実行します:

cd $SPARK_HOME
./sbin/start-thriftserver.sh --hiveconf hive.server2.thrift.port=10015

Sparkサーバーが実行中になったら、次に示すようにBeelineを起動できます:

cd $SPARK_HOME
./bin/beeline
Beeline version 1.2.1.spark2 by Apache Hive
beeline>

次に、ここで示すようにサーバーに接続します:

ノート

この例では、セキュリティを構成していないため、すべてのユーザー名とパスワードが受け入れられます。本番シナリオでは、これは行わないようにしてください。

beeline> !connect jdbc:hive2://localhost:10015 testuser testpass
Connecting to jdbc:hive2://localhost:10015
log4j:WARN No appenders could be found for logger (org.apache.hive.jdbc.Utils).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Connected to: Spark SQL (version 2.2.1)
Driver: Hive JDBC (version 1.2.1.spark2)
Transaction isolation: TRANSACTION_REPEATABLE_READ
0: jdbc:hive2://localhost:10015>

ここでどのような表が存在するかチェックすると、次のように表示されます:

0: jdbc:hive2://localhost:10015> show tables;
+-----------+------------+--------------+--+
| database  | tableName  | isTemporary  |
+-----------+------------+--------------+--+
+-----------+------------+--------------+--+
No rows selected (0.724 seconds)

この時点では何も存在していません。ただし、次に示すように、表を作成し、ダウンロードしてオブジェクト・ストレージ・バケットに配置したmovies.csvファイルにリンクできます:

0: jdbc:hive2://localhost:10015> create table test_table (movieId integer, title string, genres string) using csv options (path "oci://myBucket@myTenant/movies.csv", header "true", delimiter ",");
 
0: jdbc:hive2://localhost:10015> describe formatted test_table;
+-------------------------------+------------------------------------------------------------+----------+--+
|           col_name            |                         data_type                          | comment  |
+-------------------------------+------------------------------------------------------------+----------+--+
| movieId                       | int                                                        | NULL     |
| title                         | string                                                     | NULL     |
| genres                        | string                                                     | NULL     |
|                               |                                                            |          |
| # Detailed Table Information  |                                                            |          |
| Database                      | default                                                    |          |
| Table                         | test_table                                                 |          |
| Owner                         | opc                                                        |          |
| Created                       | Thu Mar 01 20:45:18 GMT 2018                               |          |
| Last Access                   | Thu Jan 01 00:00:00 GMT 1970                               |          |
| Type                          | EXTERNAL                                                   |          |
| Provider                      | csv                                                        |          |
| Table Properties              | [transient_lastDdlTime=1519937118]                         |          |
| Location                      | oci://PipedUploadTest@sampletenancy/movies.csv    |          |
| Serde Library                 | org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe         |          |
| InputFormat                   | org.apache.hadoop.mapred.SequenceFileInputFormat           |          |
| OutputFormat                  | org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat  |          |
| Storage Properties            | [delimiter=,, header=true, serialization.format=1]         |          |
+-------------------------------+------------------------------------------------------------+----------+--+

この表ではデータが外部のオブジェクト・ストレージに格納されることと、HDFSコネクタを使用してデータにアクセスできること(oci://ファイル・システム・スキーム)に注意してください。表が存在しているので、問合せを実行できます:

0: jdbc:hive2://localhost:10015> select * from test_table limit 10;
+----------+-------------------------------------+----------------------------------------------+--+
| movieId  |                title                |                    genres                    |
+----------+-------------------------------------+----------------------------------------------+--+
| 1        | Toy Story (1995)                    | Adventure|Animation|Children|Comedy|Fantasy  |
| 2        | Jumanji (1995)                      | Adventure|Children|Fantasy                   |
| 3        | Grumpier Old Men (1995)             | Comedy|Romance                               |
| 4        | Waiting to Exhale (1995)            | Comedy|Drama|Romance                         |
| 5        | Father of the Bride Part II (1995)  | Comedy                                       |
| 6        | Heat (1995)                         | Action|Crime|Thriller                        |
| 7        | Sabrina (1995)                      | Comedy|Romance                               |
| 8        | Tom and Huck (1995)                 | Adventure|Children                           |
| 9        | Sudden Death (1995)                 | Action                                       |
| 10       | GoldenEye (1995)                    | Action|Adventure|Thriller                    |
+----------+-------------------------------------+----------------------------------------------+--+

Oracle Cloud Infrastructureドキュメント