Spark + MaxCompute

Spark access to MaxCompute

This section introduces how to use E-MapReduce SDK to complete a MaxCompute data read/write operation in Spark.

1. Initialize an OdpsOps. In Spark, MaxCompute data operations are completed through the OdpsOps class. Create an OdpsOps object using the following stps:

        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>"
            // Taking the intranet address as an example
            val urls = Seq("http://odps-ext.aliyun-inc.com/api", "http://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))
            // The following is the calling code
            // == Step-2 ==
            ...
            // == Step-3 ==
            ...
          }
          // == Step-2 ==
          // Method definition 1
          // == Step-3 ==
          // Method definition 2
        }

2. Load table data from MaxCompute to Spark. You can load tables in MaxCompute to Spark through the readTable method of OdpsOps objects, that is, generate an RDD, as follows:

            // == 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 ==
            ...

   In the preceding code, you also need to define a read function to parse and pre-process MaxCompute table data, as shown:

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

   This function loads the first column of the MaxCompute table to the Spark running environment.

3. Save the result data in Spark to the MaxCompute table. You can store Spark RDD persistently to MaxCompute through the saveToTable method of the OdpsOps object.

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

   In the preceding code, you also need to define a write function to pre-process data before writing the data to the MaxCompute table, as shown:

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

   This function writes every line of RDD data to the first column of the corresponding MaxCompute table.

Appendix

For a complete sample code, see:

• Spark access to MaxCompute