AI 助理
文档备案控制台
官方文档
首页

模拟IDC Spark读写MaxCompute实践

更新时间：
复制 MD 格式
产品详情
我的收藏

现有湖仓一体架构是以MaxCompute为中心读写Hadoop集群数据，有些线下IDC场景，客户不愿意对公网暴露集群内部信息，需要从Hadoop集群发起访问云上的数据。本文以开源大数据开发平台E-MapReduce（云上Hadoop）方式模拟本地Hadoop集群，介绍如何读写MaxCompute数据。

背景信息

实践架构图如下所示。模拟IDC架构

准备开发环境

  • 准备E-MapReduce（EMR）环境。

    1. 购买EMR集群。

      详情请参见快速创建和使用数据湖分析集群。

    2. 登录EMR集群。

      说明

      登录E-MapReduce集群详情请参见登录集群。

      本实践登录ECS实例进行操作，连接ECS实例请参见连接ECS实例。

  • 准备本地IDEA。

    1. 安装IntelliJ IDEA。

      本实践在IntelliJ IDEA运行，需要安装IntelliJ IDEA，详情请参见Install IntelliJ IDEA。

    2. 安装Maven。

      详情请参见安装Maven。

    3. 创建Scala项目。

      1. 下载Scala插件。

        打开IDEA，选择File>Settings。在Settings对话框左侧导航栏单击Plugins，单击Scala后的Install。Scala

      2. 安装Scala JDK

      3. 创建Scala项目

        在IDEA里新建项目，选择Scala>IDEA，即可创建Scala项目。SCALA项目

  • 准备MaxCompute数据

    1. 创建项目

      MaxCompute创建Project请参见创建MaxCompute项目。

    2. 获取AccessKey

      可以进入AccessKey管理页面获取AccessKey ID和AccessKey Secret。

    3. 获取Endpoint

      MaxCompute服务的连接地址。您需要根据创建MaxCompute项目时选择的地域以及网络连接方式配置Endpoint。各地域及网络对应的Endpoint值，请参见Endpoint。

    4. 创建Table

      本实践需准备分区表和非分区表供测试使用，创建表详情请参见创建表。

读写MaxCompute数据

  1. 代码开发。

    本实践提供如下读非分区表代码开发示例。

    说明

    读分区表、写非分区表和写分区表代码示例请参见PartitionDataReaderTest.scala、DataWriterTest.scala和PartitionDataWriterTest.scala，可以根据实际业务情况进行代码开发。

    /*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.apache.spark.sql.SparkSession

/**
  * @author renxiang
  * @date 2021-12-20
  */
object DataReaderTest {

  val ODPS_DATA_SOURCE = "org.apache.spark.sql.odps.datasource.DefaultSource"
  val ODPS_ENDPOINT = "http://service.cn.maxcompute.aliyun.com/api"


  def main(args: Array[String]): Unit = {
    val odpsProject = args(0)
    val odpsAkId = args(1)
    val odpsAkKey = args(2)
    val odpsTable = args(3)

    val spark = SparkSession
      .builder()
      .appName("odps-datasource-reader")
      .getOrCreate()

    import spark._

    val df = spark.read.format(ODPS_DATA_SOURCE)
      .option("spark.hadoop.odps.project.name", odpsProject)
      .option("spark.hadoop.odps.access.id", odpsAkId)
      .option("spark.hadoop.odps.access.key", odpsAkKey)
      .option("spark.hadoop.odps.end.point", ODPS_ENDPOINT)
      .option("spark.hadoop.odps.table.name", odpsTable)
      .load()

    df.createOrReplaceTempView("odps_table")

    println("select * from odps_table")
    val dfFullScan = sql("select * from odps_table")
    println(dfFullScan.count)
    dfFullScan.show(10)

    Thread.sleep(72*3600*1000)
  }
}

  2. 代码打包和上传。

    1. Maven打包代码。

      1. 在IDEA的代码开发页面右边栏，单击Maven。

      2. 在Maven对话框，双击Lifecycle目录下的package进行打包。

    2. 本地编译jar包。

      1. 进入Project目录。

        在系统的命令行执行窗口（例如Windows的cmd窗口）执行如下命令。

        cd ${project.dir}/spark-datasource-v3.1

      2. 使用mvn命令构建spark-datasource。

        mvn clean package jar:test-jar

      3. 查看target目录下是否有dependencies.jar和tests.jar。target目录

    3. 上传jar包至服务器。

      1. 在本地使用scp命令上传已经打包好的jar包和依赖的jar包至服务器，命令语法如下。

        scp <本地jar包路径> root@<ECS实例公网IP>:<服务器存放jar包路径>

        示例如下。

        scp D:\Project\emr_mc_1\spark-datasource-v3.1\target\spark-datasource-1.0-SNAPSHOT-tests.jar root@8.xx.xx.xx:/root/emr_mc

        上传jar包

      2. 查看jar包。

        在服务器emr_mc目录下使用ll命令查看jar包。查看jar包

      3. 使用如下命令在各节点之间上传jar包。

        scp -r [本服务器存放jar包路径] root@ecs实例私网IP:[接收的服务器存放jar包地址]

  3. 运行代码。

    • 运行模式。

      • Local模式。

        • 使用Local模式运行的命令语法如下。

          ./bin/spark-submit \
    -- master local \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class DataReaderTest \
    ${jar-path} \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name}

        • 参数说明如下。

          参数

          说明

          master

          运行模式，取值如下。

          • Local：运行代码只调用当前ECS的计算资源。

          • Yarn：运行代码使用EMR集群所有ECS的计算资源，运行效率比Local模式高。

          jars

          依赖的jar包路径。

          class

          需要执行的类名称。

          jar-path

          需要执行的jar包路径。

          maxcompute-project-name

          MaxCompute的项目（Project）名称。

          aliyun-access-key-id

          阿里云账号或RAM用户的AccessKey ID。

          可以进入AccessKey管理页面获取AccessKey ID。

          aliyun-access-key-secret

          AccessKey ID对应的AccessKey Secret。

          可以进入AccessKey管理页面获取AccessKey Secret。

          maxcompute-table-name

          进行读或写的MaxCompute表名称。

      • Yarn模式。

        • 使用yarn模式运行的命令语法如下。

          val ODPS_ENDPOINT = "http://service.cn-beijing.maxcompute.aliyun-inc.com/api"

./bin/spark-submit \
    -- master yarn \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class DataReaderTest \
    ${jar-path} \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name}

        • 参数说明如下。

          参数

          说明

          master

          运行模式，取值如下。

          • Local：运行代码只调用当前ECS的计算资源。

          • Yarn：运行代码使用EMR集群所有ECS的计算资源，运行效率比Local模式高。

          jars

          依赖的jar包路径。

          class

          需要执行的类名称。

          jar-path

          需要执行的jar包路径。

          maxcompute-project-name

          MaxCompute的项目（Project）名称。

          aliyun-access-key-id

          阿里云账号或RAM用户的AccessKey ID。

          可以进入AccessKey管理页面获取AccessKey ID。

          aliyun-access-key-secret

          AccessKey ID对应的AccessKey Secret。

          可以进入AccessKey管理页面获取AccessKey Secret。

          maxcompute-table-name

          进行读或写的MaxCompute表名称。

    • 读非分区表示例。

      • 命令语法如下。

        -- 进入spark执行环境
cd /usr/lib/spark-current

-- 提交任务
./bin/spark-submit \
    -- master local \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class DataReaderTest \
    ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-tests.jar \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name}

      • 执行界面如下。读非分区表界面

      • 执行结果如下。读非分区表执行结果

    • 读分区表示例。

      • 命令语法如下。

        -- 进入spark执行环境
cd /usr/lib/spark-current

-- 提交任务
./bin/spark-submit \
    -- master local \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class PartitionDataReaderTest \
    ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-tests.jar \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name} \
    ${partition-descripion}

      • 执行界面如下。读分区表界面

      • 执行结果如下。读分区表结果

    • 写非分区表测试。

      • 命令语法如下。

        ./bin/spark-submit \
    -- master local \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class DataWriterTest \
    ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-tests.jar \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name}

      • 执行界面如下。写非分区表界面

      • 执行结果如下。写非分区表结果

    • 写分区表测试。

      • 命令语法如下。

        ./bin/spark-submit \
    -- master local \
    -- jars ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-jar-with-dependencies.jar,${project.dir}/spark-datasource-v2.3/libs/cupid-table-api-1.1.5-SNAPSHOT.jar,${project.dir}/spark-datasource-v2.3/libs/table-api-tunnel-impl-1.1.5-SNAPSHOT.jar \
    -- class PartitionDataWriterTest \
    ${project.dir}/spark-datasource-v3.1/target/spark-datasource-1.0-SNAPSHOT-tests.jar \
    ${maxcompute-project-name} \
    ${aliyun-access-key-id} \
    ${aliyun-access-key-secret} \
    ${maxcompute-table-name} \
    ${partition-descripion}

      • 执行界面如下。写分区表界面

      • 执行结果如下。写分区表结果

性能测试

本实践性能测试环境是E-MapReduce和MaxCompute，属于云上互联。如果IDC网络与云上相连性能取决于tunnel资源或者专线带宽。

  • 实例规格。

    实例

    规格

    E-MapReduce集群

    • Master节点数量：2个。

      • ECS规格：计算型（ecs.c6.2xlarge）8 vCPU，16 GiB，2.5 Gbps。

      • 系统盘：ESSD云盘 120GiB。

      • 数据盘：ESSD云盘 80GiB。

    • Core节点数量：2个。

      • ECS规格：计算型（ecs.c6.2xlarge）8 vCPU，16 GiB，2.5 Gbps。

      • 系统盘：ESSD云盘 120GiB。

      • 数据盘：ESSD云盘 80GiB * 4。

    MaxCompute

    按量计费标准版。

  • 大表读测试。

    数据表规格如下。

    参数

    规格

    表名称

    dwd_product_movie_basic_info

    说明

    此表为MaxCompute公开数据集MAXCOMPUTE_PUBLIC_DATA项目下的表，详情请参见公开数据集。

    表大小

    4829258484 Byte。

    分区数

    593 个。

    读取的分区名称

    20170422。

    结果如下。大表读结果耗时 0.850871秒。

  • 大表写测试。

    • 分区写入万条数据。分区写入万条数据耗时2.533892秒。

    • 分区写入十万条数据。分区写入十万条数据耗时8.441193秒。

    • 分区写入百万条数据。分区写入百万条数据耗时73.28秒。

上一篇：PyODPS节点实现避免将数据下载到本地下一篇：Spark on MaxCompute访问Phoenix数据