E-MapReduce中Spark 2.x读写MaxCompute数据

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 最新的`aliyun-emapreduce-sdk`将`MaxCompute`数据以`DataSource`的方式接入Spark 2.x,用户可以使用类似Spark 2.x中读写`json/parquet/csv`的方式来访问MaxCompute.

最新的aliyun-emapreduce-sdkMaxCompute数据以DataSource的方式接入Spark 2.x,用户可以使用类似Spark 2.x中读写json/parquet/csv的方式来访问MaxCompute.

0. DataSource

a) DataSource提供了一种插件式的外部数据接入SparkSQL的方式,数据源只要实现相应的DataSource API即可以整合进SparkSQL,它的特点如下:

  • 通过DataSet/DataFrame/sparkSQLText等标准方式来访问数据源
  • SparkSQL引擎优化
  • scala语言接入后,Spark支持的其它语言也可以进行访问,如pyspark等

Spark 2.x内置支持的数据源:

  • json
  • csv
  • parquet
  • orc
  • text
  • jdbc

Spark 2.x 访问数据源示例:

b) 读数据

  val df = spark.read.json("pathToJson")
  
  // 提供schema信息
  val schemaType = StructType(Seq(
    StructField("year", IntegerType, nullable = true))
  spark.read.schema(schemaType)json("pathToJson")

  // 带一些参数设置,如csv的分隔符等
  spark.read.option("header", "false").option("sep", ",").csv("pathToCsv"")
  
  // load api, 等同于spark.read.json("pathToLoad")
  spark.read.format("json").load("pathToLoad")
  
  // sql方式访问
  df.createOrReplaceTempView("t")
  spark.sql("select * from t")

c) 写数据

  val df = Seq(1, 2, 3).toDF("a")
  df.write.json("jsonWritePath")
  
  // 等同上面写法
  df.write.format("json").save("jsonWritePath")

  
  // 带参数
  df.write
    .option("header", "true")
    .option("compression", "gZiP")
    .csv("csvWritePath")

  // 路径存在,则覆盖
  df.write.mode("overwrite").json("jsonWritePath")

d)sparkSQLText using DataSource

   spark.sql("create table t(a string) using json")
   spark.sql("insert int table t select 1")
   spark.sql("select * from t")
   ...

1. MaxCompute以DataSource接入Spark 2.x

如上介绍了DataSource的特点以及读写方式,MaxCompute作为一个数据源,通过E-MapReduce的aliyun-emapreduce-sdk也可以通过上述方式来访问。

1.1. aliyun-emapreduce-sdk

Git地址

branch: master-2.x

1.2 SparkSQL读写MaxCompute

a) option参数设置

访问MaxCompute表中的数据,需要一些参数,如下:

parameter optional value
odpsUrl No 内网地址: http://odps-ext.aliyun-inc.com/api 公网地址:http://service.odps.aliyun.com/api
tunnelUrl No 内网地址: http://dt-ext.odps.aliyun-inc.com 公网地址:http://dt.odps.aliyun.com
accessKeySecret No 阿里云accessKeySecret
accessKeyId No 阿里云accessKeyId
project No MaxCompute项目空间
table No MaxCompute表名
numPartitions Yes 表的Partition个数,默认 1
partitionSpec Yes 分区信息,如pt=xxx,多个用逗号分开pt=xxx,dt=xxx
allowCreatNewPartition Yes 分区不存在是否创建,默认 false
b) 写数据
  • MaxCompute中必须已经存在表()),若没有需要去MaxCompute控制台进行创建
  • DataFrame中的数据写入MaxCompute的表中
val df = Seq(("Hello", "E-MapReduce")).toDF("a","b")
df.write.format("org.apache.spark.aliyun.maxcompute.datasource")
  .option("odpsUrl", "http://odps-ext.aliyun-inc.com/api")
  .option("tunnelUrl", "http://dt-ext.odps.aliyun-inc.com")
  .option("table", "t")
  .option("project", "test_odpss")
  .option("accessKeyId", "your accessKeyId")
  .option("accessKeySecret", "your accessKeySecret")
  .mode("overwrite") //覆盖写
  .save()
  
case class MyClass(a: String, b: String)
val df1 = Seq(MyClass("Hello", "World")).toDF
df1.write.format("org.apache.spark.aliyun.maxcompute.datasource")
  .option("odpsUrl", "http://odps-ext.aliyun-inc.com/api")
  .option("tunnelUrl", "http://dt-ext.odps.aliyun-inc.com")
  .option("table", "t")
  .option("project", "test_odpss")
  .option("accessKeyId", "your accessKeyId")
  .option("accessKeySecret", "your accessKeySecret")
  .mode("append") //append追加
  .save()
  
  
// 写分区表 建表:create table t1(a string) partitioned by(b string)
val df2 = Seq("E-MapReduce").toDF("a") // 不包含分区列
df2.write.format("org.apache.spark.aliyun.maxcompute.datasource")
  .option("odpsUrl", "http://odps-ext.aliyun-inc.com/api")
  .option("tunnelUrl", "http://dt-ext.odps.aliyun-inc.com")
  .option("table", "t1")
  .option("project", "test_odpss")
  .option("partitionSpec","b='Hello'") // 分区描述
  .option("allowCreatNewPartition","true") //若分区不存在,是否创建
  .option("accessKeyId", "your accessKeyId")
  .option("accessKeySecret", "your accessKeySecret")
  .mode("append") //append追加
  .save()
  

备注:

DataFrame的列名和类型必须和MaxCompute的表中一致

不支持spark.write.parititonBy

不支持动态分区

  • MaxCompute控制台查询表数据进行验证
序号 a b
1 Hello E-MapReduce
2 Hello World
c) 读数据
  • 从上述表中读取数据到DataFrame
val df = spark
  .read
  .format("org.apache.spark.aliyun.maxcompute.datasource")
  .option("odpsUrl", "http://odps-ext.aliyun-inc.com/api")
  .option("tunnelUrl", "http://dt-ext.odps.aliyun-inc.com")
  .option("table", "t")
  .option("project", "test_odpss")
  .option("accessKeyId", "your accessKeyId")
  .option("accessKeySecret", "your accessKeySecret")
  .load()

df.show(false)

+-----+-----------+
|a    |b          |
+-----+-----------+
|Hello|E-MapReduce|
|Hello|World      |
+-----+-----------+

// 读出为DataFrame后可进行DataFrame的各种操作,如join
val df1 = Seq(("Hello", "AliYun")).toDF("a", "c")
df.join(df1, "a").show(false)

+-----+-----------+-------+
|a    |b          |c      |
+-----+-----------+-------+
|Hello|E-MapReduce|AliYun|
|Hello|World      |AliYun|
+-----+-----------+-------+

// 也可注册为Spark的临时表
df.createOrReplaceTempView("test_t")
spark.sql("select * from test_t").show(false)

df1.createOrReplaceTempView("test_t_1")
spark.sql("select * from test_t join test_t_1 on test_t.a = test_t_1.a ")


// 读分区表 建表:create table t2(a string) partitioned by(b string)
spark.read.format("org.apache.spark.aliyun.maxcompute.datasource")
  .option("odpsUrl", "http://odps-ext.aliyun-inc.com/api")
  .option("tunnelUrl", "http://dt-ext.odps.aliyun-inc.com")
  .option("table", "t2") // table t2
  .option("project", "test_odpss")
  .option("partitionSpec","b='Hello'") // 分区描述
  .option("accessKeyId", "your accessKeyId")
  .option("accessKeySecret", "your accessKeySecret")
  .save()
  
+-----------+
|a          |
+-----------+
|E-MapReduce|
+-----------+
d) sparkSQLText
  • 不支持在sparkSQLText直接对MaxCompute表进行相关操作
  • 可以通过上述读数据的方式使用DataFrame注册成临时表的方式,进行相关操作(insert不支持)
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
目录
相关文章
|
12小时前
|
分布式计算 运维 Serverless
E-MapReduce Serverless Spark开发者评测
**EMR Serverless Spark测评概要** - 弹性处理大规模用户行为分析,提升产品优化与推荐精度。 - 相比自建Spark集群,EMR Serverless Spark展现更高稳定性、性能,降低成本,简化运维。 - 支持多种数据源,提供Spark SQL与DataFrame API,自动资源调度,适用于波动需求。 - 文档清晰,但可增强特定场景指导与故障排查。 - 建议优化监控、调度算法,增加内置分析工具,并强化与其他阿里云产品(如MaxCompute, DataWorks, QuickBI)的联动。 - 全托管服务减轻运维负担,但资源管理、查询效率与兼容性仍有提升空间。
7 0
|
1天前
|
分布式计算 数据处理 流计算
实时计算 Flink版产品使用问题之使用Spark ThriftServer查询同步到Hudi的数据时,如何实时查看数据变化
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2天前
|
SQL 分布式计算 Serverless
E-MapReduce Serverless Spark 评测
E-MapReduce Serverless Spark 评测
|
9天前
|
SQL 分布式计算 Java
E-MapReduce Serverless Spark体验评测
从了解到部署实践,全方位带你体验大数据平台EMR Serverless Spark的魅力。
134 7
E-MapReduce Serverless Spark体验评测
|
18天前
|
弹性计算 分布式计算 DataWorks
DataWorks产品使用合集之spark任务如何跨空间取表数据
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
12 1
|
1月前
|
SQL 分布式计算 HIVE
实时计算 Flink版产品使用问题之同步到Hudi的数据是否可以被Hive或Spark直接读取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
分布式计算 定位技术 Scala
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
|
1月前
|
存储 分布式计算 Hadoop
Hadoop生态系统详解:HDFS与MapReduce编程
Apache Hadoop是大数据处理的关键,其核心包括HDFS(分布式文件系统)和MapReduce(并行计算框架)。HDFS为大数据存储提供高容错性和高吞吐量,采用主从结构,通过数据复制保证可靠性。MapReduce将任务分解为Map和Reduce阶段,适合大规模数据集的处理。通过代码示例展示了如何使用MapReduce实现Word Count功能。HDFS和MapReduce的结合,加上YARN的资源管理,构成处理和分析大数据的强大力量。了解和掌握这些基础对于有效管理大数据至关重要。【6月更文挑战第12天】
52 0
|
1月前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
26 1
|
1月前
|
分布式计算 Java Hadoop
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
26 0