开发者社区> huangchuang> 正文

OSS数据湖实践——EMR + Flink + OSS案例

简介: 构建基于OSS数据源的EMR大数据计算环境,使用Flink大数据计算引擎,实现简单的大数据分析案例。
+关注继续查看

本文介绍使用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,运行作业

1589441726617_2f91171e_9a01_404a_ac74_2e998d9c3d2d

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

步骤五:查看作业运行是否成功及查看运行结果

1589442550574_5b094f4d_cbc1_4c40_baea_b875bff45021
1589442588976_7872ee42_c97a_4082_a071_974c02006c98

总结

通过以上步骤,可以了解spark 处理OSS数据源的整个过程,这将对后续其他任务作业开发带来初步的参考。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
18715 0
数据湖实操讲解【OSS 访问加速】第十讲:Impala 如何高效查询 OSS 数据
数据湖 JindoFS+OSS 实操干货 36讲 每周二16点准时直播! 扫文章底部二维码入钉群,线上准时观看~ Github链接: https://github.com/aliyun/alibabacloud-jindofs
3013 0
十大行业经典案例!Apache Flink 的 40 个最佳实践
如今,Apache Flink 行业应用几何?在降本增效的需求驱动下,企业如何实现数据与算力价值最大化?本文整理了 Flink 社区近一年的社区案例,并按照行业进行分类,供大家参考!
7481 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
23687 0
联合利华搭上阿里云数据中台 精准营销新客提升2倍
联合利华中国数据与数字化发展副总裁方军说,“过去,品牌全域拉新运营就像是盲盒,但现在我们有了全域数字化抓手。”打通多平台,为品牌天猫业务进一步开展随时提供基础数据服务,阿里云数据中台能够为包括联合利华旗下品牌在内的诸多品牌,带来非常大的价值。
2128 0
Flex borderSkin绘自定义宽度边框的实例
<p>项目打印需要,要有边框,而且要设置边框的宽度。我们项目是用Twaver flex做的,经过查资料和尝试,采用在Network的rootCanvas上绘制边框。但Canvas的样式有限,不能直接修改边框宽度。经过一番尝试,通过borderSkin样式能达到我要的效果。由于在网上没有找到类似的文章,所以自己记录下,方便自己查阅,以及和大家分享讨论下这个方法。</p> <p>这是我自定义
1183 0
+关注
6
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载