E-MapReduce Spark + ODPS是什么?

Spark + MaxCompute



Spark 接入 MaxCompute


本章节将介绍如何使用E-MapReduce SDK在Spark中完成一次MaxCompute数据的读写操作。

1. 
   初始化一个OdpsOps对象。在 Spark中，MaxCompute的数据操作通过OdpsOps类完成，请参照如下步骤创建一个OdpsOps对象：
2.      import com.aliyun.odps.TableSchema
3.      import com.aliyun.odps.data.Record
4.      import org.apache.spark.aliyun.odps.OdpsOps
5.      import org.apache.spark.{SparkContext, SparkConf}
7.      object Sample {
8.        def main(args: Array[String]): Unit = {    
9.          // == Step-1 ==
10.          val accessKeyId = "<accessKeyId>"
11.          val accessKeySecret = "<accessKeySecret>"
12.          // 以内网地址为例
13.          val urls = Seq("http://odps-ext.aliyun-inc.com/api", "http://dt-ext.odps.aliyun-inc.com")
15.          val conf = new SparkConf().setAppName("Test Odps")
16.          val sc = new SparkContext(conf)
17.          val odpsOps = OdpsOps(sc, accessKeyId, accessKeySecret, urls(0), urls(1))
19.          // 下面是一些调用代码
20.          // == Step-2 ==
21.          ...
22.          // == Step-3 ==
23.          ...
24.        }
26.        // == Step-2 ==
27.        // 方法定义1
28.        // == Step-3 ==
29.        // 方法定义2
30.      ｝

从MaxCompute中加载表数据到Spark中。通过OdpsOps对象的readTable方法，可以将MaxCompute中的表加载到Spark中，即生成一个RDD，如下所示：

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

在上面的代码中，您还需要定义一个read函数，用来解析和预处理MaxCompute表数据，如下所示：

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

这个函数的含义是将MaxCompute表的第一列加载到Spark运行环境中。
将 Spark 中的结果数据保存到MaxCompute表中。通过OdpsOps对象的saveToTable方法，可以将SparkRDD持久化到MaxCompute中。

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

在上面的代码中，您还需要定义一个write函数，用作写MaxCompute表前数据预处理，如下所示：

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

这个函数的含义是将RDD的每一行数据写到对应MaxCompute表的第一列中。
分区表参数写法说明
SDK支持对MaxCompute分区表的读写，这里分区名的写法标准是：分区列名=分区名，多个分区时以逗号分隔，例如有分区列pt和ps：

• 读分区pt为1的表数据：pt=‘1’
• 读分区pt为1和分区ps为2的表数据：pt=‘1’,ps=‘2’

附录


示例代码请看:

• Spark接入MaxCompute