本文介绍使用Flink大数据分析引擎，基于EMR，利用OSS云存储数据，实现一个分析案例。
前提条件
• 已注册阿里云账号，详情请参见注册云账号。
• 已开通E-MapReduce服务和OSS服务。
• 已完成云账号的授权，详情请参见角色授权。
• 已创建Haoop集群，且带有spark组件。
• 相关更多配置请参考OSS入门文档。
步骤一：数据上传至oss
hadoop fs -put course2.csv oss://your-bucket-name/步骤二：编写处理代码，及打包
package org.myorg.quickstart
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.TableEnvironment
object OSSExample {
def main(args: Array[String]) {
// set up the batch execution environment
case class Course(Id : Int, Subject : String, Level : String)
val env = ExecutionEnvironment.getExecutionEnvironment
val tableEnv = BatchTableEnvironment.create(env)
val data: DataSet[(Long, String, String)] = env.readCsvFile("oss://your-bucket-name/course.csv")
val course = tableEnv.fromDataSet[(Long, String, String)](data, 'id, 'subject, 'level)
val counts = course.groupBy("subject, level").select("subject, level, level.count as cnt")
val maxcounts = counts.groupBy("subject").select("subject as subject1, cnt.max as cnt1")
val result = maxcounts.leftOuterJoin(counts, "cnt=cnt1").select("subject, level, cnt")
result.toDataSet[(String, String, Long)].print()
}
}IDEA Build -> Build Artifact ->Build 打包为OSSFlinkExample jar包
步骤三：上传jar包到Hadoop 或者OSS
把jar 上传到集群header节点，然后使用以下命令
hadoop fs -put OSSExample.jar oss://your-bucket-name/
步骤四：创建FLink作业job，运行作业
run -m yarn-cluster -yjm 1024 -ytm 1024 -yn 4 -ys 4 -ynm flink-oss-sample -c org.myorg.quickstart.OSSExample ossref://your-bucket-name/OSSFlinkExample.jar
步骤五：查看作业运行是否成功及查看运行结果
总结
通过以上步骤，可以了解spark 处理OSS数据源的整个过程，这将对后续其他任务作业开发带来初步的参考。
