Oracle Cloud Infrastructureドキュメント

SparkでのHDFSコネクタの使用

概要

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

前提条件

チュートリアルを完了するための前提条件は次のとおりです:

  • コンピュート・インスタンスを起動する権限が必要です。 ガイダンスについては、「インスタンスの起動」を参照してください。
  • 起動したサービス・インスタンスに接続できる必要があります。 ガイダンスについては、「インスタンスへの接続」を参照してください。
  • オブジェクト・ストレージと対話するために使用するアイデンティティおよびアクセス管理(IAM)ユーザーには、適切なOCID、フィンガープリント、および秘密キーが必要です。 ガイダンスについては、「SDKとツールの構成」を参照してください。「リソース識別子」も参照してください。
  • 接続できるオブジェクト・ストレージ・バケットが必要です。
  • IAMユーザーは、コンソールを使用してそのバケットに読み書きすることができなければなりません。

スパークの使用

スパークと依存関係のインストール

ノート

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

ノート

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

  1. コンピュート・サービスのインスタンスを起動します。 ガイダンスについては、「インスタンスの起動」を参照してください。
  2. Secure Shell (SSH)接続を使用して接続できるように、サービス・インスタンスにパブリックIPアドレスがあることを確認します。 ガイダンスについては、「インスタンス・コンソール接続」を参照してください。
  3. SSH接続を使用してサービス・インスタンスに接続します。
  4. 以下のコード例を使用して、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://confluence.oci.oraclecorp.com/pages/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.cloud.oracle.com/Content/API/SDKDocs/hdfsconnector.htm#two
# 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データセットを使用します。

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

スパーク・シェルを使用したテスト

データが準備できたら、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
 
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サーバーを起動

Spark Thrift Serverをポート10015で起動し、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                    |
+----------+-------------------------------------+----------------------------------------------+--+

詳細情報