PolarDB是阿里云自研的下一代关系型云数据库,100%兼容MySQL,适用于企业多样化的数据库应用场景。本文主要介绍如何通过DLA Serverless Spark访问云数据库PolarDB。

前提条件

  • 已经开通对象存储OSS(Object Storage Service)服务。具体操作请参考开通OSS服务
  • 已经创建云数据库PolarDB MySQL实例。具体请参考数据库
  • 在PolarDB MySQL实例中已创建数据表,并插入数据。参考命令样例如下:
    #建表语句:
    CREATE TABLE `testdb`.`test_table` (
     `name` varchar(32) NULL,
     `age` INT  NULL,
     `score` DOUBLE NULL
    )
    #插入数据语句:
    INSERT INTO `testdb`.`test_table` VALUES('aliyun01', 1001, 10.1);
    INSERT INTO `testdb`.`test_table` VALUES('aliyun02', 1002, 10.2);
    INSERT INTO `testdb`.`test_table` VALUES('aliyun03', 1003, 10.3);
    INSERT INTO `testdb`.`test_table` VALUES('aliyun04', 1004, 10.4);
    INSERT INTO `testdb`.`test_table` VALUES('aliyun05', 1005, 10.5);
  • 准备DLA Spark访问PolarDB MySQL实例所需的安全组ID和交换机ID。具体操作请参见配置数据源网络
  • DLA Spark访问PolarDB MySQL实例所需的安全组ID或交换机ID,已添加到PolarDB MySQL实例所在集群的白名单中。具体操作请参见设置集群白名单

操作步骤

  1. 准备以下测试代码和依赖包来访问PolarDB MySQL,并将此测试代码和依赖包分别编译打包生成jar包上传至您的OSS。
    测试代码示例:
    package com.aliyun.spark
    
    import java.util.Properties
    import org.apache.spark.sql.SparkSession
    
    object SparkOnPOLARDB {
    
      def main(args: Array[String]): Unit = {
        //获取POLARDB的url、database、tableName、登录POLARDB数据库的user和password。
        val url = args(0)
        val jdbcConnURL = s"jdbc:mysql://$url"
        val database = args(1)
        val tableName = args(2)
        val user = args(3)
        val password = args(4)
    
        //Spark侧的表名。
        var sparkTableName = args(5)
    
        val sparkSession = SparkSession
          .builder()
          .appName("scala spark on POLARDB test")
          .getOrCreate()
    
        val driver = "com.mysql.cj.jdbc.Driver"
    
        //如果存在的话就删除表。
        sparkSession.sql(s"drop table if exists $sparkTableName")
    
        //Sql方式,Spark会映射POLARDB中表的Schema。
        val createCmd =
          s"""CREATE TABLE ${sparkTableName} USING org.apache.spark.sql.jdbc
             |    options (
             |    driver '$driver',
             |    url '$jdbcConnURL',
             |    dbtable '$database.$tableName',
             |    user '$user',
             |    password '$password'
             |    )""".stripMargin
        println(s"createCmd: \n $createCmd")
        sparkSession.sql(createCmd)
        val querySql = "select * from " + sparkTableName + " limit 1"
        sparkSession.sql(querySql).show
    
        //使用dataset API接口。
        val connectionProperties = new Properties()
        connectionProperties.put("driver", driver)
        connectionProperties.put("user", user)
        connectionProperties.put("password", password)
        //读取数据。
        var jdbcDf = sparkSession.read.jdbc(jdbcConnURL,
          s"$database.$tableName",
          connectionProperties)
        jdbcDf.select("name", "age", "score").show()
    
        val data =
          Seq(
            PersonPolardb("bill", 30, 170.5),
            PersonPolardb("gate", 29, 200.3)
          )
        val dfWrite = sparkSession.createDataFrame(data)
    
        //写入数据。
        dfWrite
          .write
          .mode("append")
          .jdbc(jdbcConnURL, s"$database.$tableName", connectionProperties)
        jdbcDf.select("name", "age").show()
        sparkSession.stop()
      }
    }
    
    case class PersonPolardb(name: String, age: Int, score: Double)
    PolarDB MySQL依赖的pom文件:
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.22</version>
            </dependency>
  2. 登录Data Lake Analytics管理控制台
  3. 在页面左上角,选择PolarDB MySQL实例所在地域。
  4. 单击左侧导航栏中的Serverless Spark > 作业管理
  5. 作业编辑页面,单击创建作业
  6. 创建作业页面,按照页面提示进行参数配置后,单击确定创建Spark作业。
    3
  7. 单击Spark作业名,在Spark作业编辑框中输入以下作业内容,并按照以下参数说明进行参数值替换。保存并提交Spark作业。
    {
        "args": [
            "pc-xxx.mysql.polardb.rds.aliyuncs.com:3306",  #POLARDB的私网地址和端口。
            "testdb",  #POLARDB中的数据库名。
            "test_table",  #POLARDB中的表名。
            "xxx1",  #POLARDB中的数据库登录的用户名。
            "xxx2",  #POLARDB中的数据库登录的用户名。
            "spark_on_polardb_table"  #Spark中创建映射POLARDB表的表名。
        ],
        "file": "oss://spark_test/jars/polardb/spark-examples-0.0.1-SNAPSHOT.jar",  #存放测试代码的OSS路径。
        "name": "polardb-test",
        "jars": [
            "oss://spark_test/jars/polardb/mysql-connector-java-8.0.22.jar"  #存放测试代码依赖包的OSS路径。
        ],
        "className": "com.aliyun.spark.SparkOnPOLARDB",
        "conf": {
            "spark.driver.resourceSpec": "small",
            "spark.executor.instances": 2,
            "spark.executor.resourceSpec": "small",
            "spark.dla.eni.enable": "true",
            "spark.dla.eni.vswitch.id": "vsw-xxx",   #可访问PolarDB的交换机id。
            "spark.dla.eni.security.group.id": "sg-xxx"  #可访问PolarDB的安全组id。
        }
    }

执行结果

作业运行成功后,在任务列表中单击操作 > 日志,查看作业日志。出现如下日志说明作业运行成功:日志详情