E-MapReduce:Spark を使用して OSS にアクセスする

Spark Resilient Distributed Dataset (RDD) を使用して OSS からデータを読み取り、OSS にデータを書き込む

次の例は、AccessKey ペアを指定せずに、Spark を使用して OSS からデータを読み取り、処理済みデータを OSS に書き戻す方法を示しています。

1. SSH モードでクラスタのマスターノードにログオンします。詳細については、「クラスタのマスターノードにログオンする」をご参照ください。

2. 次のコマンドを実行して、Spark Shell を起動します。

   spark-shell

3. ビジネス要件に基づいて次のコードのパラメーターを変更した後、Spark Shell で次の Scala コードを実行して、OSS からデータを読み取り、OSS にデータを書き込みます。

   import org.apache.spark.{SparkConf, SparkContext}
   val conf = new SparkConf().setAppName("Test OSS")
   val sc = new SparkContext(conf)
   val pathIn = "oss://<yourBucket>/path/to/read"
   val inputData = sc.textFile(pathIn)
   val cnt = inputData.count
   println(s"count: $cnt")
   val outputPath = "oss://<yourBucket>/path/to/write"
   val outputData = inputData.map(e => s"$e has been processed.")
   outputData.saveAsTextFile(outputPath)

   この例では、ビジネス要件に基づいて次のパラメーターを置き換えます。

   • yourBucket: OSS バケットの名前を指定します。

   • pathIn: データを読み取るファイルのパスを指定します。

   • outputPath: データを書き込むファイルのパスを指定します。

   完全なサンプルコードについては、GitHub にアクセスしてください。

PySpark を使用して OSS からデータを読み取り、OSS にデータを書き込む

次の例は、PySpark を使用して OSS からデータを読み取り、処理済みデータを AccessKey ペアを指定せずに OSS に書き戻す方法を示しています。

1. SSH モードでクラスタのマスターノードにログオンします。詳細については、「クラスタのマスターノードにログオンする」をご参照ください。

2. 次のコマンドを実行して、PySpark のインタラクティブ環境に移動します。

   pyspark

3. ビジネス要件に基づいて次のコードのパラメーターを変更した後、PySpark でコードを実行します。

   from pyspark.sql import SparkSession
   spark = SparkSession.builder.appName("Python Spark SQL OSS example").getOrCreate()
   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)
   

   この例では、ビジネス要件に基づいて次のパラメーターを置き換えます。

   • yourBucket: OSS バケットの名前を指定します。

   • pathIn: データを読み取るファイルのパスを指定します。

   • outputPath: データを書き込むファイルのパスを指定します。

Spark SQL ステートメントを実行して CSV ファイルを作成し、OSS に保存する