Spark SQL_开发指南_Serverless Spark_云原生数据湖分析 DLA

Spark跟DLA SQL引擎以及数据湖构建服务共享元数据。

Spark访问数据湖元数据服务

Spark引擎可以支持多种元数据服务，既支持访问用户自建的HiveMetaStore（参考文档：Hive数据源连接），也支持跟访问DLA统一的数据湖元数据。DLA统一的数据湖元数据管理服务，同时支持多种引擎访问，实现多种引擎的元数据信息共享。在数据湖执行DLA SQL、元数据爬取、一键建仓中创建的库表结构, 可以被Spark读取并使用，Spark SQL创建或者修改的元数据也可以被其他引擎访问到。下图是Spark SQL 和 DLA SQL与元数据服务之间的关系。

您可以在DLA控制台左侧的SQL执行处看到所有数据湖的中的数据库和表结构，对表进行在线分析查询，以及管理子用户对库表的权限，如下图所示。

提交Spark SQL作业

在DLA Spark控制台，如果需要连接DLA元数据服务, 只需要在提交作业时, 加入配置项:"spark.sql.hive.metastore.version": "dla" 。以下示例是将数据湖元数据中1k_tables 数据库下的table0 表的内容提取出来:

{
    "sqls": [
        "select * from `1k_tables`.`table0` limit 100",
        "insert into `1k_tables`.`table0` values(1, 'test')"
    ],
    "name": "sql test",
    "jars": [
        "oss://test/hive-serde-3.1.2.jar"
    ],
    "conf": {
        "spark.dla.connectors": "oss",
        "spark.driver.resourceSpec": "small",
        "spark.sql.hive.metastore.version": "dla",
        "spark.sql.catalogImplementation": "hive",
        "spark.executor.instances": 10,
        "spark.dla.job.log.oss.uri": "oss://test/spark-logs",
        "spark.executor.resourceSpec": "small"
    }
}

说明：

"sqls"关键字允许用户不提交jar包，直接提交SQL作业，使用方式更简单，适合于熟悉SQL语言的开发者。参数值是一个数组，可以在一个作业中执行多条SQL语句，如上所示，不同的语句用逗号隔开。
除了控制台，您也可以通过OpenAPI提交Spark SQL作业，提交方式跟其他类型Spark作业相同。
我们也提供了 spark-sql工具包来提交sql作业，具体请参考
Spark-SQL工具。

代码中使用Spark SQL

用户也可以在程序中执行SQL，操作元数据信息，以及读写表内容。下面以PySpark为例进行介绍，其他语言使用方式类似。首先，建立以下的Python文件，保存为example.py，将文件上传至oss。

from pyspark.sql import SparkSession

if __name__ == "__main__":
    # init pyspark context
    spark = SparkSession \
        .builder \
        .appName("Python SQL Test") \
        .getOrCreate()
    
    # create a database
    spark.sql(
            "create database if not exists dlatest comment 'c' location 'oss://test/warehouse/' WITH DBPROPERTIES(k1='v1', k2='v2')")
    # create table
    spark.sql(
            "create table dlatest.tp(col1 INT)  PARTITIONED BY (p1 STRING, p2 STRING) location 'oss://test/warehouse/tp' STORED AS parquet TBLPROPERTIES ('parquet.compress'='SNAPPY')")
    # show structure
    print(spark.sql("show create table tp").collect()[0])
    
    # insert data
    spark.sql("INSERT into tp partition(p1='a',p2='a') values(1)")
    
    # show data
    spark.sql("select * from tp").show()

通过以下的JSON将作业通过DLA Spark控制台提交

{
    "name": "DLA SQL Test",
    "file": "oss://path/to/example.py"
    "conf": {
        "spark.driver.resourceSpec": "small",
        "spark.sql.hive.metastore.version": "dla",
        "spark.sql.catalogImplementation": "hive",
        "spark.dla.connectors": "oss",
        "spark.executor.instances": 1,
        "spark.dla.job.log.oss.uri": "oss://path/to/spark-logs",
        "spark.executor.resourceSpec": "small"
    }
}

执行成功后, 可以在数据湖控制台的SQL执行页面中找到一个名为dlatest的数据库，以及其下的tp表。

注意

数据湖元数据服务对命名大小写不敏感，在引用库名和表名时忽略大小写。

读写Hive存储格式的表时的注意事项

当使用一键建仓服务, 或者在控制台手动执行SQL的方式创建表，并指定表的数据读写格式为Hive某些格式，比如JSON、CSV时，需要在提交Spark作业时, 引入相关的Jar包。

注意

推荐您在Maven官方仓库下载Hive Serde的Jar包，如hive-serde

当提交Spark作业时，如果表的数据存储格式是Hive的某种格式，需要将对应的Jar包作为第三方Jar提交, 如下所示

{
    "name": "DLA Meta Test",
    "sqls": ["SELECT * FROM HiveDB.JsonTable"],
    "jars": [
        "oss://test/hive-serde-3.1.2.jar"
    ],
    "conf": {
        "spark.driver.resourceSpec": "small",
        "spark.sql.hive.metastore.version": "dla",
        "spark.sql.catalogImplementation": "hive",
        "spark.dla.connectors": "oss",
        "spark.executor.instances": 1,
        "spark.dla.job.log.oss.uri": "oss://test/spark-logs",
        "spark.executor.resourceSpec": "small"
    }
}

使用限制和注意事项

1.当前在Spark中仅支持External类型数据库和表的创建和读写操作

当前Spark连接数据湖元数据服务，只支持外表(External Table)的读写和创建操作。

这意味着在建立数据库时, 需要显式指定LOCATION信息，类似如下的SQL语句

CREATE DATABASE db1 LOCATION 'oss://test/db1/'

同样的,建表语句必须显式指定表的存储LOCATION信息, 类似如下SQL语句

CREATE TABLE table1(col1 INT) LOCATION 'oss://test/db1/table1/'

需要注意以下几个事项

当用户在Spark中DROP一个表或者表的某个PARTITION时，并不会删除OSS上的文件。
当用户创建一个表时，指定的表的LOCATION必须是库的LOCATION的子文件夹。
当用户为添加表的PARTITION 时，指定的PARTITION LOCATION必须是表的LOCATION的子文件夹。
当用户RENAME PARTITION` 时，并不会改变其在OSS上的路径结构。

2. 当前Spark中仅支持以OSS为存储的外表

在当前阶段，数据湖分析SQL执行支持多种不同的存储，包括RDS、表格存储等等。当前在Spark中使用元数据服务支持读写以OSS为存储的外表。

使用Spark直接创建数据库和数据表，LOCATION必须指定一个合法的OSS地址。

对于其他存储的支持，后续会陆续更新。

3. 当前禁止创建DEFAULT为名字的数据库

由于不允许使用DEFAULT为名字的数据库，需要注意以下两个事项

禁止在Spark中创建和操作名为DEFAULT的数据库。
在Spark中执行SQL时, 操作表之前需要使用USE DatabaseNameSQL语句来切换到目标数据。或者显式指定某个表属于哪个数据库, 例如SELELCT * FROM db1.table1。

4. 当前在Spark中执行 ALTER 语句有部分限制

用户可以通过类似ALTER DATABASE ... 等语句修改库和表的元数据信息，当前在Spark中使用这一类语句有如下的限制。

当前对数据库，仅支持修改COMMENT信息，其它如LOCATION、PROPERTIES禁止修改。
当前仅支持修改表的COLUMN 和PROPERTIES信息，如添加列、修改注解等等。请注意这里的COLUMN必须是非PARTITION列。

5. 当前Spark不支持在元数据服务中存储UDF信息

当前用户依然可以在Spark代码中创建和使用UDF, 但这些UDF无法被注册到元数据服务器成为通用的函数，您可以在每个作业中先创建临时函数，然后在该作业中使用该函数。如下：

CREATE TEMPORARY FUNCTION IF NOT EXISTS func1 as
'com.aliyun.dla.SimpleUDFExample' USING JAR '/path/to/jar1',
JAR '/path/to/jar2'

持久化函数会在后续的版本中支持。

6. 当前在 SQL执行 中操作Spark表的一些限制

当用户在Spark程序中创建了数据库db1和表table1后，如果在控制台的SQL执行中操作它们，需要显式检测其是否存在。如尝试删除此数据库时必须使用如下语句

DROP DATABASE IF EXISTS db1

7. 当前在Spark SQL中不支持GRANT类赋权语句

同开源社区一致，当前用户无法通过Spark引擎执行一个GRANT语句来修改子账户赋权，需要修改子用户权限可以在数据湖分析的执行SQL页面上通过执行GRANT/REVOKE语句的方式修改用户的权限。