E-MapReduce:Spark を OSS に接続する

Spark Shell (Scala) の使用

1. SSH 経由でマスターノードにログインします。

2. Spark Shell を起動します。

   spark-shell

3. 以下のコードを実行します。<yourBucket> は、ご利用の OSS バケット名に置き換えてください。

   scala> val pathIn = "oss://<yourBucket>/path/to/read"
   scala> val inputData = sc.textFile(pathIn)
   scala> val cnt = inputData.count
   cnt: Long = ...
   scala> println(s"count: $cnt")
   scala> val outputPath = "oss://<yourBucket>/path/to/write"
   scala> val outputData = inputData.map(e => s"$e has been processed.")
   scala> outputData.saveAsTextFile(outputPath)

   このコードは入力パスからすべての行を読み取り、行数をカウントし、各行にサフィックスを追加して、結果を出力パスに書き込みます。完全なサンプルについては、GitHub 上の「SparkOssDemo.scala」をご参照ください。

PySpark の使用

1. SSH 経由でマスターノードにログインします。

2. PySpark を起動します。

   pyspark

3. 以下のコードを実行します。<yourBucket> は、ご利用の OSS バケット名に置き換えてください。

   >>> pathIn = "oss://<yourBucket>/path/to/read"
   >>> df = spark.read.text(pathIn)
   >>> cnt = df.count()
   >>> print(cnt)
   >>> outputPath = "oss://<yourBucket>/path/to/write"
   >>> df.write.format("parquet").mode("overwrite").save(outputPath)

   このコードは入力パスをテキスト形式の DataFrame として読み取り、行数を表示し、結果を Parquet フォーマットで出力パスに書き込みます。

Spark SQL の使用

1. SSH 経由でマスターノードにログインします。

2. Spark SQL CLI を起動します。

   spark-sql

3. OSS に保存されるデータベースを作成し、CSV テーブルを作成して行を挿入します。

   パラメーター	説明
   delimiter	CSV ファイルでフィールドを区切る文字。
   header	最初の行に列名が含まれる場合は true、そうでない場合は false を設定します。

   CREATE DATABASE test_db LOCATION "oss://<yourBucket>/test_db";
   USE test_db;
   CREATE TABLE student (id INT, name STRING, age INT)
       USING CSV OPTIONS ("delimiter"=";", "header"="true");
   INSERT INTO student VALUES(1, "ab", 12);
   SELECT * FROM student;

   <yourBucket> は、ご利用の OSS バケット名に置き換えてください。SELECT 文の実行結果は以下のようになります。

   1    ab    12

4. 結果を確認するには、OSS 内の CSV ファイルを確認します。ファイルはセミコロンで区切られており、最初の行がヘッダーになっています。

   id;name;age
   1;ab;12

手順 1：パスワードなしの構成を削除する

Hadoop-Common サービスの core-site.xml ファイルから fs.oss.credentials.provider パラメーターを削除します。

手順 2：パスワードなしのアクセスが削除されたことを確認する

以下のコマンドを実行します。

hadoop fs -ls oss://<yourBucket>/test_db

削除に成功すると、以下が表示されます。

ls: ERROR: not found login secrets, please configure the accessKeyId and accessKeySecret.

手順 3：AccessKey パラメーターを core-site.xml に追加する

Hadoop-Common サービスの core-site.xml ファイルに、以下のパラメーターを追加します。

手順 4：AccessKey 構成を確認する

以下のコマンドを実行します。

hadoop fs -ls oss://<yourBucket>/test_db

AccessKey ペアが正しく構成されている場合、OSS パスの一覧が表示されます。

drwxrwxrwx   - root root          0 2025-02-24 11:45 oss://<yourBucket>/test_db/student

手順 5：Spark サービスを再起動する

すべての Spark 関連サービスを再起動します。サービスが実行中になったら、Spark Shell、PySpark、または Spark SQL を使用して OSS の読み取りおよび書き込みを行ってください。