Spark SQL

Spark访问数据湖元数据服务

Spark引擎可以支持多种元数据服务，既支持访问用户自建的Hive，也支持访问DLA统一管理的数据湖元数据。DLA统一管理的数据湖元数据管理服务，同时支持多种引擎访问，实现多种引擎的元数据信息共享。在数据湖元信息发现、T+1全量同步一键建仓中创建的库表结构, 可以被Spark读取并使用，Spark SQL创建或者修改的元数据也可以被其他引擎访问到。下图是Spark SQL和DLA SQL与元数据服务之间的关系。

纯SQL作业

DLA Spark支持直接在控制台写Spark SQL。无需用户打包jar包或者写python代码，更有利于数据开发人员使用Spark进行数据分析。

您需要先登录DLA控制台，在Serverless Spark > 作业管理菜单中创建SparkSQL类型的作业。创建SparkSQL类型的作业后，系统默认会使用DLA元数据服务。如果您想关闭DLA元数据服务，可以使用以下两种方式：

• 使用in-memory catalog，将不会使用DLA元数据服务。

  set spark.sql.catalogImplementation = in-memory;

• 设置hive metastore version为1.2.1或其他版本。

  set spark.sql.catalogImplementation = hive;
  set spark.sql.hive.metastore.version = 1.2.1;

SparkSQL作业的文本框中，支持直接写SQL语句，每条SQL语句以分号隔开。

SQL语句支持下列类型命令：

• SET命令

• 用于指定Spark的设置，一般置于整个SQL语句的最前面。

• 每条Set命令指定一个Spark参数的值，每条SET命令用分号隔开。

• SET命令的Key和Value均不要加单引号或者双引号。

• ADD JAR命令

• 用于增加Spark SQL运行时，依赖的jar包，比如UDF的jar包，各类数据源连接器的Jar包等。Jar包目前支持OSS格式路径，一般置于整个SQL语句的最前面。

• 每条add jar命令指定一个oss jar包路径，路径字符串不要加单引号和双引号， 每条ADD JAR命令用分号隔开。

• Spark SQL语法所支持的DDL或DML语句

• 例如查询语句select。

• 例如插入语句insert。

• 例如查看数据库SHOW DATABASE。

代码中使用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://{your bucket name}/{path}/' WITH DBPROPERTIES(k1='v1', k2='v2')")
    # create table
    spark.sql(
            "create table dlatest.tp(col1 INT)  PARTITIONED BY (p1 STRING, p2 STRING) location 'oss://{your bucket name}/{path}/' STORED AS parquet TBLPROPERTIES ('parquet.compress'='SNAPPY')")
    # show structure
    print(spark.sql("show create table dlatest.tp").collect()[0])
    
    # insert data
    spark.sql("INSERT into dlatest.tp partition(p1='a',p2='a') values(1)")
    
    # show data
    spark.sql("select * from dlatest.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"
    }
}

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

使用DLA元数据服务的限制和注意事项

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

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

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为存储的外表。

• 禁止在Spark中创建和操作名为DEFAULT的数据库。

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

• 当前对数据库，仅支持修改COMMENT信息，其它如LOCATION、PROPERTIES禁止修改。

• 当前仅支持修改表的COLUMN 和PROPERTIES信息，如添加列、修改注解等等。请注意这里的COLUMN必须是非PARTITION列。

DROP DATABASE IF EXISTS db1;