X-Pack Spark 访问OSS-阿里云开发者社区

开发者社区> 云计算> 正文
登录阅读全文

X-Pack Spark 访问OSS

简介: 简介 对象存储服务(Object Storage Service,OSS)是一种海量、安全、低成本、高可靠的云存储服务,适合存放任意类型的文件。容量和处理能力弹性扩展,多种存储类型供选择,全面优化存储成本。

简介

对象存储服务(Object Storage Service,OSS)是一种海量、安全、低成本、高可靠的云存储服务,适合存放任意类型的文件。容量和处理能力弹性扩展,多种存储类型供选择,全面优化存储成本。
本文主要介绍通过Spark操作OSS数据的常见方式,代码以Scala为例。本文的代码可以通过“数据工作台”提交。

前置条件

  1. OSS已经创建bucket,假设名称为:test_spark
  2. 已创建具备读写OSS bucket:test_spark权限的用户。假设用户名为test_oss,访问OSS的AccessKeyID和AccessKeySecret分别为:accessId,accessKey。
  3. OSS的路径格式为:oss://${AccessKeyID}:${AccessKeySecret}@${bucketName}.${endPoint}/${ossKeyPath}。例如:oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/spark-table/test.csv

使用Spark读写OSS文件样例

假设有如下内容的文本数据已经存在OSS中,路径为:oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/spark-table/test.csv ,内容为:

101, name_101, 0.52
102, name_102, 0.78
103, name_103, 0.76
104, name_104, 0.78
105, name_105, 0.02
106, name_106, 0.29
107, name_107, 0.63
108, name_108, 0.20
109, name_109, 0.07
110, name_110, 0.33

通过Spark读取文件,常用两种方法

一、 使用DataFrame 读取,实例代码如下:

val conf = new SparkConf().setAppName("spark sql test")
val sparkSession = SparkSession
      .builder()
      .config(conf)
      .enableHiveSupport()
      .getOrCreate()
val ossCsvPath = s"oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/spark-table/test.csv"
//读取test.csv并生产DataFrame
val fileDF = sparkSession.read.csv(ossCsvPath)
//打印fileDF内容
fileDF.show()
//也可以把fileDF 注册是Spark表
fileDF.createOrReplaceTempView(“test_table")
sparkSession.sql("select * from test_table").show()    

二、 创建Spark Sql表指向test.csv,实例代码如下:

val sql =
      s"""create table test_table(
         |      id          int,
         |      name        string,
         |      value       float
         |      ) row format delimited fields terminated by ','
         |      location 'oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/spark-table/'
         |      """.stripMargin
//创建spark 表
sparkSession.sql(sql)
//查询表数据
sparkSession.sql("select * from test_table").show()

通过Spark写文件,常用DataFrame写文件。

示例代码如下:

val conf = new SparkConf().setAppName("spark sql test")
val sparkSession = SparkSession
      .builder()
      .config(conf)
      .enableHiveSupport()
      .getOrCreate()
val ossCsvPath = s"oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/spark-table/test.csv"
//读取test.csv并生产DataFrame
val fileDF = sparkSession.read.csv(ossCsvPath)
//打印fileDF内容
fileDF.show()
val writeOssParquetPath = "oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/parquet-table/"
//写parquet格式文件
fileDF.write.parquet(writeOssParquetPath)
val writeCsvParquetPath = "oss://accessId:accessKey@test_spark.oss-cn-shenzhen-internal.aliyuncs.com/user/csv-table/"
//写csv格式文件
fileDF.write.csv(writeCsvParquetPath)

小结

本文给出Spark操作OSS数据的基本用法,更多用法会陆续推出。

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

分享:
云计算
使用钉钉扫一扫加入圈子
+ 订阅

时时分享云计算技术内容,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。

其他文章
最新文章
相关文章