全部产品

Spark SQL

更新时间:2020-11-09 10:46:31

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

Spark访问数据湖元数据服务

Spark引擎可以支持多种元数据服务,既支持访问用户自建的HiveMetaStore(参考文档:Hive数据源连接),也支持跟访问DLA统一的数据湖元数据。DLA统一的数据湖元数据管理服务,同时支持多种引擎访问,实现多种引擎的元数据信息共享。在数据湖执行DLA SQL、元数据爬取、一键建仓中创建的库表结构, 可以被Spark读取并使用,Spark SQL创建或者修改的元数据也可以被其他引擎访问到。下图是Spark SQL 和 DLA SQL与元数据服务之间的关系。
数据湖多引擎连接元数据管理
您可以在DLA控制台左侧的SQL执行处看到所有数据湖的中的数据库和表结构,对表进行在线分析查询,以及管理子用户对库表的权限,如下图所示。
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"
    }
}
说明:
  1. "sqls"关键字允许用户不提交jar包,直接提交SQL作业,使用方式更简单,适合于熟悉SQL语言的开发者。参数值是一个数组,可以在一个作业中执行多条SQL语句,如上所示,不同的语句用逗号隔开。

  2. 除了控制台,您也可以通过OpenAPI提交Spark SQL作业,提交方式跟其他类型Spark作业相同。

  3. 我们也提供了 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信息,其它如LOCATIONPROPERTIES禁止修改。

  • 当前仅支持修改表的COLUMNPROPERTIES信息,如添加列、修改注解等等。请注意这里的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语句的方式修改用户的权限。