アプリケーションからのデータへのアクセス

外部ボリューム内でのファイルの処理

Analytics Engine powered by Apache Spark を使用して実行される Spark アプリケーションでは、 Spark ジョブ ファイル、入力データ、または出力データを参照する一般的な方法は、 IBM Cloud Pak for Data Volumes API を使用して管理できる外部ストレージ・ボリュームを介して行うことです。

以下の外部ボリュームを処理できます。

外部 NFS ストレージ・ボリューム

外部 NFS サーバーでのボリュームの作成 の外部 NFS ・サーバーでボリュームを作成する方法については、こちらを参照してください。

既存の永続ボリューム要求

永続ボリューム要求でのボリュームの作成 を参照してください。

新規ボリューム・インスタンス

永続ボリューム・インスタンスの管理 を参照してください。

ボリューム・インスタンスを使用してディレクトリーの作成およびアプリケーション・ファイルの追加を行う方法については、『ボリューム API を使用した永続ボリューム・インスタンスの管理』を参照してください。

複数のストレージ・ボリューム内でのファイルの処理

Spark ジョブ・ペイロードを作成するときに、複数のストレージ・ボリュームを使用できます。

以下の例は、Spark クラスター上に /myapp としてマウントされる vol1 ボリューム内の customApps ディレクトリーの下にアップロードされる Spark アプリケーションを示しています。 ユーザー・データは、Spark クラスター上に /data としてマウントされる vol2 ボリューム内にあります。

{
  "application_details": {
    "application": "/myapp/<spark_application>",
    "arguments": [
      ""
    ],
    "conf": {
      "spark.app.name": "JSFVT",
      "spark.executor.extraClassPath": "/myapp/*",
      "spark.driver.extraClassPath": "/myapp/*"
    }
  },
  "volumes": [
    {
      "name": "<project_name>::vol1",
      "mount_path": "/myapp",
      "source_sub_path": "customApps"
    },
    {
      "name": "<project_name>::vol2",
      "source_sub_path": "",
      "mount_path": "/data"
    }
  ]
}

オブジェクト・ストレージでのファイルの処理

ジョブ・ファイルとデータを S3 互換のオブジェクト・ストレージ・バケットに保管できます。 以下のステップでは、IBM Cloud Object Storage バケットでこれを行う方法を説明します。

• 以下のように、アプリケーション (例えば、Python プログラム・ファイル cosExample.py) を作成します。

from __future__ import print_function

import sys
import calendar
import time

from pyspark.sql import SparkSession
if __name__ == "__main__":
    if len(sys.argv) != 5:
        print("Usage: cosExample <access-key> <secret-key> <endpoint> <bucket>", file=sys.stderr)
        sys.exit(-1)

    spark = SparkSession.builder.appName("CosExample").getOrCreate()

    prefix = "fs.cos.llservice"
    hconf = spark.sparkContext._jsc.hadoopConfiguration()
    hconf.set(prefix +".endpoint", sys.argv[3])
    hconf.set(prefix + ".access.key", sys.argv[1])
    hconf.set(prefix + ".secret.key", sys.argv[2])

    data = [1, 2, 3, 4, 5, 6]
    distData = spark.sparkContext.parallelize(data)

    distData.count()
    path = "cos://{}.llservice/{}".format(sys.argv[4], calendar.timegm(time.gmtime()))
    distData.saveAsTextFile(path)

    rdd = spark.sparkContext.textFile(path)
    print ("output rdd count: {}". format(rdd.count()))
    spark.stop()

• ジョブ・ファイルをロードします。 外部ボリュームからジョブ・ファイルをロードするには、Spark クラスター内に /myapp としてマウントされるストレージ・ボリューム vol1 内の customApps ディレクトリーの下に cosExample.py をアップロードします。

{
    "application_details": {
    "application": "/myapp/cosExample.py",
    "arguments": ["<ACCESS_KEY>", "<COS_SECRET_KEY>", "<COS_ENDPOINT>", "<BUCKET_NAME>"],
    "class": "org.apache.spark.deploy.SparkSubmit",
    "conf": {
        "spark.app.name": "Job1",
        "spark.executor.extraClassPath": "/myapp/*",
        "spark.driver.extraClassPath": "/myapp/*"
    }
    },
    "volumes": [{
        "name": "<project_name>::vol1",
        "mount_path": "/myapp",
        "source_sub_path": "customApps"
        }]
}

• あるいは、IBM Cloud Object Storage バケットからジョブ・ファイルをロードするには、IBM Cloud Object Storage サービス (<COS_SERVICE_NAME>) 内のバケット <BUCKET_NAME> から <OBJECT_NAME> にジョブ・ファイルをアップロードします。

{
    "application_details": {
        "application": "cos://<BUCKET_NAME>.<COS_SERVICE_NAME>/<OBJECT_NAME>",
        "arguments": ["cos://<BUCKET_NAME>.<COS_SERVICE_NAME>/<OBJECT_NAME>"],
        "class": "<main_class>",
        "conf": {
            "spark.app.name": "MyJob",
            "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.endpoint": "<COS_ENDPOINT>",
            "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.secret.key": "<COS_SECRET_KEY>",
            "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.access.key": "<COS_ACCESS_KEY>"
        }
    }
}