精选30+云产品，助力企业轻松上云！>>>

本文介绍使用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数据源的整个过程，这将对后续其他任务作业开发带来初步的参考。