836      技術社區[雲棲]

Spark + ODPS__Spark_開發人員指南_E-MapReduce-阿裏雲

Spark + MaxCompute

Spark 接入 MaxCompute

本章節將介紹如何使用E-MapReduce SDK在Spark中完成一次MaxCompute數據的讀寫操作。

1. 初始化一個OdpsOps對象。在 Spark 中，MaxCompute的數據操作通過OdpsOps類完成，請參照如下步驟創建一個OdpsOps對象：

        import com.aliyun.odps.TableSchema
        import com.aliyun.odps.data.Record
        import org.apache.spark.aliyun.odps.OdpsOps
        import org.apache.spark.{SparkContext, SparkConf}
        object Sample {
          def main(args: Array[String]): Unit = {    
            // == Step-1 ==
            val accessKeyId = "<accessKeyId>"
            val accessKeySecret = "<accessKeySecret>"
            // 以內網地址為例
            val urls = Seq("https://odps-ext.aliyun-inc.com/api", "https://dt-ext.odps.aliyun-inc.com") 
            val conf = new SparkConf().setAppName("Test Odps")
            val sc = new SparkContext(conf)
            val odpsOps = OdpsOps(sc, accessKeyId, accessKeySecret, urls(0), urls(1))
            // 下麵是一些調用代碼
            // == Step-2 ==
            ...
            // == Step-3 ==
            ...
          }
          // == Step-2 ==
          // 方法定義1
          // == Step-3 ==
          // 方法定義2
        ｝

2. 從MaxCompute中加載表數據到Spark中。通過OdpsOps對象的readTable方法，可以將MaxCompute中的表加載到Spark中，即生成一個RDD，如下所示：

            // == Step-2 ==
            val project = <odps-project>
            val table = <odps-table>
            val numPartitions = 2
            val inputData = odpsOps.readTable(project, table, read, numPartitions)
            inputData.top(10).foreach(println)
            // == Step-3 ==
            ...

   在上麵的代碼中，您還需要定義一個read函數，用來解析和預處理MaxCompute表數據，如下所示：

            def read(record: Record, schema: TableSchema): String = {
              record.getString(0)
            }

   這個函數的含義是將MaxCompute表的第一列加載到Spark運行環境中。

3. 將 Spark 中的結果數據保存到MaxCompute表中。通過OdpsOps對象的saveToTable方法，可以將Spark RDD持久化到MaxCompute中。

            val resultData = inputData.map(e => s"$e has been processed.")
            odpsOps.saveToTable(project, table, dataRDD, write)

   在上麵的代碼中，您還需要定義一個write函數，用作寫MaxCompute表前數據預處理，如下所示：

            def write(s: String, emptyReord: Record, schema: TableSchema): Unit = {
              val r = emptyReord
              r.set(0, s)
            }

   這個函數的含義是將RDD的每一行數據寫到對應MaxCompute表的第一列中。

4. 分區表參數寫法說明

SDK支持對MaxCompute分區表的讀寫，這裏分區名的寫法標準是：分區列名=分區名，多個分區時以逗號分隔，例如有分區列pt和ps：