DataWorks_数据开发_EMR Spark节点_计算Pi和对接MaxCompute案例

/** Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements.  See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License.  You may obtain a copy of the License at**    http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/packagecom.aliyun.emr.example.sparkimportcom.aliyun.odps.TableSchemaimportcom.aliyun.odps.data.Recordimportorg.apache.spark.aliyun.odps.OdpsOpsimportorg.apache.spark.{SparkConf, SparkContext}
objectSparkMaxComputeDemo {
defmain(args: Array[String]): Unit= {
if (args.length<6) {
System.err.println(
"""Usage: SparkMaxComputeDemo <accessKeyId> <accessKeySecret> <envType> <project> <table> <numPartitions>||Arguments:
||accessKeyIdAliyunAccessKeyID.
|accessKeySecretAliyunKeySecret.
|envType0or1|0: Publicenvironment.
|1: Aliyuninternalenvironment, i.e. AliyunECSetc.
|projectAliyunODPSproject|tableAliyunODPStable|numPartitionsthenumberofRDDpartitions""".stripMargin)System.exit(1)
    }
valaccessKeyId=args(0)
valaccessKeySecret=args(1)
valenvType=args(2).toIntvalproject=args(3)
valtable=args(4)
valnumPartitions=args(5).toIntvalurls=Seq(
Seq("http://service.odps.aliyun.com/api", "http://dt.odps.aliyun.com"), // public environmentSeq("http://odps-ext.aliyun-inc.com/api", "http://dt-ext.odps.aliyun-inc.com") // Aliyun internal environment    )
valconf=newSparkConf().setAppName("E-MapReduce Demo 3-1: Spark MaxCompute Demo (Scala)")
valsc=newSparkContext(conf)
valodpsOps=envTypematch {
case0=>OdpsOps(sc, accessKeyId, accessKeySecret, urls(0)(0), urls(0)(1))
case1=>OdpsOps(sc, accessKeyId, accessKeySecret, urls(1)(0), urls(1)(1))
    }
valodpsData=odpsOps.readTable(project, table, read, numPartitions)
println(s"Count (odpsData): ${odpsData.count()}")
  }
defread(record: Record, schema: TableSchema): Long= {
record.getBigint(0)
  }
}