常用概念 客户端 客户端直接面向用户，可通过Java API、Thrift API访问服务端进行Hive的相关操作。 HQL语言 Hive Query Language，类SQL语句。 HCatalog HCatalog是建立在Hive元数据之上的一个表信息管理层，吸收了Hive的DDL命令。为MapReduce提供读写接口，提供Hive命令行接口来进行数据定义和元数据查询。基于 MRS 的HCatalog功能，Hive、MapReduce开发人员能够共享元数据信息，避免中间转换和调整，能够提升数据处理的效率。 WebHCat WebHCat运行用户通过Rest API来执行Hive DDL，提交MapReduce任务，查询MapReduce任务执行结果等操作。

Hive介绍 Hive是一个开源的，建立在Hadoop上的 数据仓库 框架，提供类似SQL的HQL语言操作结构化数据，其基本原理是将HQL语言自动转换成MapReduce任务或Spark任务，从而完成对Hadoop集群中存储的海量数据进行查询和分析。 Hive主要特点如下： 通过HQL语言非常容易的完成数据提取、转换和加载（ETL）。 通过HQL完成海量结构化数据分析。 灵活的数据存储格式，支持JSON、 CS V、TEXTFILE、RCFILE、ORCFILE、SEQUENCEFILE等存储格式，并支持自定义扩展。 多种客户端连接方式，支持JDBC接口。 Hive主要应用于海量数据的离线分析（如 日志分析 ，集群状态分析）、大规模的数据挖掘（用户行为分析，兴趣分区，区域展示）等场景下。 为保证Hive服务的高可用性、用户数据的安全及访问服务的可控制，在开源社区的Hive-3.1.0版本基础上，Hive新增如下特性： 数据文件加密机制。 开源社区的Hive特性，请参见https://cwiki.apache.org/confluence/display/hive/designdocs。

IoTDB样例工程介绍 MRS样例工程获取地址为https://github.com/huaweicloud/huaweicloud-mrs-example，切换分支为与MRS集群相匹配的版本分支，然后下载压缩包到本地后解压，即可获取各组件对应的样例代码工程。 当前MRS提供以下IoTDB相关样例工程： 表1 IoTDB相关样例工程 样例工程位置 描述 iotdb-examples/iotdb-flink-example 通过Flink访问IoTDB数据的示例程序，包括FlinkIoTDBSink和FlinkIoTDBSource。 FlinkIoTDBSink可实现通过Flink job将时序数据写入到IoTDB中。FlinkIoTDBSource则通过Flink job将时序数据从IoTDB读取出来并且打印，，相关样例介绍请参见IoTDB Flink样例程序。 iotdb-examples/iotdb-jdbc-example IoTDB JDBC处理数据Java示例程序。 本示例演示了如何使用JDBC接口连接IoTDB，并执行IoTDB SQL语句，相关样例介绍请参见IoTDB JDBC样例程序。 iotdb-examples/iotdb-kafka-example 通过Kafka访问IoTDB数据的示例程序。 本示例演示了如何先将时序数据发送到Kafka，再使用多线程将数据写入到IoTDB中，相关样例介绍请参见IoTDB Kafka样例程序。 iotdb-examples/iotdb-session-example IoTDB Session处理数据Java示例程序。 本示例演示了如何使用Session方式连接IoTDB，并执行IoTDB SQL语句，相关样例介绍请参见IoTDB Session样例程序。 iotdb-examples/iotdb-udf-exmaple 该样例程序介绍了如何实现一个简单的IoTDB自定义函数（UDF），相关样例介绍请参见IoTDB自定义函数（UDF）样例程序。

常用概念 以电力场景为例，说明如何在IoTDB中创建一个正确的数据模型。 图1 电力场景属性层级组织结构 如图1所示，即“电力集团层-电厂层-设备层-传感器层”。其中ROOT为根节点，传感器层的每一个节点为叶子节点。IoTDB的语法规定，ROOT节点到叶子节点的路径以“.”连接，以此完整路径命名IoTDB中的一个时间序列。例如，图1最左侧路径对应的时间序列名称为“ROOT.ln.wf01.wt01.status”。 基本概念： 设备 设备指的是在实际场景中拥有传感器的装置。在IoTDB当中，所有的传感器都应有其对应的归属的设备。 传感器 传感器是指在实际场景中的一种检测装置，它能感受到被测量的信息，并能将感受到的信息按一定规律变换成为电信号或其他所需形式的信息输出并发送给IoTDB。在IoTDB当中，存储的所有的数据及路径，都是以传感器为单位进行组织。 存储组 用户可以将任意前缀路径设置成存储组。如果有4条时间序列，如“root.vehicle.d1.s1”、“root.vehicle.d1.s2”、“root.vehicle.d2.s1”和“root.vehicle.d2.s2”，路径“root.vehicle”下的两个设备d1、d2可能属于同一个业主或者同一个厂商，因此关系紧密。这时候就可以将前缀路径“root.vehicle”指定为一个存储组，这将使得IoTDB将其下的所有设备的数据存储在同一个文件夹下。如果以后“root.vehicle”路径下增加了新的设备，也将属于该存储组。 设置合理数量的存储组可以带来性能的提升。既不会因为产生过多的存储文件（夹）导致频繁切换IO降低系统速度（并且会占用大量内存且出现频繁的内存-文件切换），也不会因为过少的存储文件夹（降低了并发度）导致写入命令阻塞。 用户应根据自己的数据规模和使用场景，平衡存储文件的存储组设置，以达到更好的系统性能。 时间序列 时间序列是IoTDB中的核心概念。时间序列可以被看作产生时序数据的传感器所在的完整路径，在IoTDB中所有的时间序列必须以root开始、以传感器作为结尾。

Hive应用开发流程介绍 开发流程中各阶段的说明如图1和表1所示。 图1 Hive应用程序开发流程 表1 Hive应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解Hive的基本概念。 常用概念 准备开发和运行环境 Hive的应用程序支持使用Java、Python两种语言进行开发。推荐使用IntelliJ IDEA工具，请根据指导完成不同语言的开发环境配置。 准备Hive应用开发环境 根据场景开发工程 提供了Java、Python两种不同语言的样例工程，还提供了从建表、数据加载到数据查询的样例工程。 开发Hive应用 运行程序及结果查看 指导用户将开发好的程序编译提交运行并查看结果。 调测Hive应用 父主题： Hive开发指南（安全模式）

常见问题 问题一：查询OBS分区表报错，报错信息如下： DLI .0005: There should be at least one partition pruning predicate on partitioned table `xxxx`.`xxxx`.; 问题根因：查询OBS分区表时没有携带分区字段。 解决方案：查询OBS分区表时，where条件中至少包含一个分区字段。 问题二：使用DataSource语法指定OBS文件路径创建OBS表，insert数据到OBS表，显示作业运行失败，报：“DLI.0007: The output path is a file, don't support INSERT...SELECT” 错误。 问题示例语句参考如下： CREATE TABLE testcsvdatasource (name string, id int) USING csv OPTIONS (path "obs://dli-test-021/data/test.csv"); 问题根因：创建OBS表指定的OBS路径为具体文件，导致不能插入数据。例如上述示例中的OBS路径为："obs://dli-test-021/data/test.csv"。 解决方案：使用DataSource语法创建OBS表指定的OBS文件路径改为文件目录即可，后续即可通过insert插入数据。上述示例，建表语句可以修改为： CREATE TABLE testcsvdatasource (name string, id int) USING csv OPTIONS (path "obs://dli-test-021/data"); 问题三：使用Hive语法创建OBS分区表时，提示语法格式不对。例如，如下使用Hive语法创建以classNo为分区的OBS表： CREATE TABLE IF NOT EXISTS testtable(name STRING, score DOUBLE, classNo INT) PARTITIONED BY (classNo) STORED AS TEXTFILE LOCATION 'obs://dli-test-021/data7'; 问题根因：使用Hive语法创建OBS分区表时，分区字段不能出现在表名后的字段列表中，只能定义在PARTITIONED BY后。 解决方案：使用Hive语法创建OBS分区表时，分区字段指定在PARTITIONED BY后。例如： CREATE TABLE IF NOT EXISTS testtable(name STRING, score DOUBLE) PARTITIONED BY (classNo INT) STORED AS TEXTFILE LOCATION 'obs://dli-test-021/data7';

DataSource和Hive两种语法创建OBS表的区别 DataSource语法和Hive语法主要区别在于支持的表数据存储格式范围、支持的分区数等有差异。两种语法创建OBS表主要差异点参见表1。 表1 DataSource语法和Hive语法创建OBS表的差异点 语法 支持的数据类型范围 创建分区表时分区字段差异 支持的分区数 DataSource语法 支持ORC，PARQUET，JSON，CSV，AVRO类型 创建分区表时，分区字段在表名和PARTITIONED BY后都需要指定。具体可以参考DataSource语法创建单分区OBS表。 单表分区数最多允许7000个。 Hive语法 支持TEXTFILE, AVRO, ORC, SEQUENCEFILE, RCFILE, PARQUET 创建分区表时，指定的分区字段不能出现在表后，只能通过PARTITIONED BY指定分区字段名和类型。具体可以参考Hive语法创建OBS分区表。 单表分区数最多允许100000个。 创建OBS表的DataSource语法可以参考使用DataSource语法创建OBS表。 创建OBS表的Hive语法可以参考使用Hive语法创建OBS表。

使用DataSource语法创建OBS表 以下通过创建CSV格式的OBS表举例，创建其他数据格式的OBS表方法类似，此处不一一列举。 创建OBS非分区表 指定OBS数据文件，创建csv格式的OBS表。 按照以下文件内容创建“test.csv”文件，并将“test.csv”文件上传到OBS桶“dli-test-021”的根目录下。 Jordon,88,23 Kim,87,25 Henry,76,26 登录DLI管理控制台，选择“SQL编辑器”，在SQL编辑器中“执行引擎”选择“spark”，“队列”选择已创建的SQL队列，数据库选择“testdb”，执行以下命令创建OBS表。 CREATE TABLE testcsvdatasource (name STRING, score DOUBLE, classNo INT ) USING csv OPTIONS (path "obs://dli-test-021/test.csv"); 如果是通过指定的数据文件创建的OBS表，后续不支持在DLI通过insert表操作插入数据。OBS文件内容和表数据保持同步。 查询已创建的“testcsvdatasource”表数据。 select * from testcsvdatasource; 图1 查询结果 本地修改原始的OBS表文件“test.csv”，增加一行“Aarn,98,20”数据，重新替换OBS桶目录下的“test.csv”文件。 Jordon,88,23 Kim,87,25 Henry,76,26 Aarn,98,20 在DLI的SQL编辑器中再次查询“testcsvdatasource”表数据，DLI上可以查询到新增的“Aarn,98,20”数据。 select * from testcsvdatasource; 图2 查询结果 指定OBS数据文件目录，创建csv格式的OBS表。 指定的OBS数据目录不包含数据文件。 在OBS桶“dli-test-021”根目录下创建数据文件目录“data”。 登录DLI管理控制台，选择“SQL编辑器”，在SQL编辑器中“执行引擎”选择“spark”，“队列”选择已创建的SQL队列，数据库选择“testdb”。在DLI的“testdb”数据库下创建OBS表“testcsvdata2source”。 CREATE TABLE testcsvdata2source (name STRING, score DOUBLE, classNo INT) USING csv OPTIONS (path "obs://dli-test-021/data"); 通过insert语句插入表数据。 insert into testcsvdata2source VALUES('Aarn','98','20'); insert作业运行成功后，查询OBS表“testcsvdata2source”数据。 select * from testcsvdata2source; 图3 查询结果 在OBS桶的“obs://dli-test-021/data”目录下刷新后查询，生成了csv数据文件，文件内容为insert插入的数据内容。 图4 查询结果 指定的OBS数据目录包含数据文件。 在OBS桶“dli-test-021”根目录下创建数据文件目录“data2”。创建如下内容的测试数据文件“test.csv”，并上传文件到“obs://dli-test-021/data2”目录下。 Jordon,88,23 Kim,87,25 Henry,76,26 登录DLI管理控制台，选择“SQL编辑器”，在SQL编辑器中“执行引擎”选择“spark”，“队列”选择已创建的SQL队列，数据库选择“testdb”。在DLI的“testdb”数据库下创建OBS表“testcsvdata3source”。 CREATE TABLE testcsvdata3source (name STRING, score DOUBLE, classNo INT) USING csv OPTIONS (path "obs://dli-test-021/data2"); 通过insert语句插入表数据。 insert into testcsvdata3source VALUES('Aarn','98','20'); insert作业运行成功后，查询OBS表“testcsvdata3source”数据。 select * from testcsvdata3source; 图5 查询结果 在OBS桶的“obs://dli-test-021/data2”目录下刷新后查询，生成了一个csv数据文件，内容为insert插入的表数据内容。 图6 查询结果 创建OBS分区表 创建单分区OBS表 在OBS桶“dli-test-021”根目录下创建数据文件目录“data3”。 登录DLI管理控制台，选择“SQL编辑器”，在SQL编辑器中“执行引擎”选择“spark”，“队列”选择已创建的SQL队列，数据库选择“testdb”。在DLI的“testdb”数据库下创建以“classNo”列为分区的OBS分区表“testcsvdata4source”，指定OBS目录“obs://dli-test-021/data3”。 CREATE TABLE testcsvdata4source (name STRING, score DOUBLE, classNo INT) USING csv OPTIONS (path "obs://dli-test-021/data3") PARTITIONED BY (classNo); 在OBS桶的“obs://dli-test-021/data3”目录下创建“classNo=25”的分区目录。根据以下文件内容创建数据文件“test.csv”，并上传到OBS的“obs://dli-test-021/data3/classNo=25”目录下。 Jordon,88,25 Kim,87,25 Henry,76,25 在SQL编辑器中执行以下命令，导入分区数据到OBS表“testcsvdata4source ”。 ALTER TABLE testcsvdata4source ADD PARTITION (classNo = 25) LOCATION 'obs://dli-test-021/data3/classNo=25'; 查询OBS表“testcsvdata4source ”classNo分区为“25”的数据： select * from testcsvdata4source where classNo = 25; 图7 查询结果 插入如下数据到OBS表“testcsvdata4source ”： insert into testcsvdata4source VALUES('Aarn','98','25'); insert into testcsvdata4source VALUES('Adam','68','24'); 查询OBS表“testcsvdata4source ”classNo分区为“25”和“24”的数据。 分区表在进行查询时where条件中必须携带分区字段，否则会查询失败，报：DLI.0005: There should be at least one partition pruning predicate on partitioned table。 select * from testcsvdata4source where classNo = 25; 图8 查询结果 select * from testcsvdata4source where classNo = 24; 图9 查询结果 在OBS桶的“obs://dli-test-021/data3”目录下点击刷新，该目录下生成了对应的分区文件，分别存放新插入的表数据。 图10 OBS上classNo分区为“25”文件数据 图11 OBS上classNo分区为“24”文件数据 创建多分区OBS表 在OBS桶“dli-test-021”根目录下创建数据文件目录“data4”。 登录DLI管理控制台，选择“SQL编辑器”，在SQL编辑器中“执行引擎”选择“spark”，“队列”选择已创建的SQL队列，数据库选择“testdb”。在“testdb”数据库下创建以“classNo”和“dt”列为分区的OBS分区表“testcsvdata5source”，指定OBS目录“obs://dli-test-021/data4”。 CREATE TABLE testcsvdata5source (name STRING, score DOUBLE, classNo INT, dt varchar(16)) USING csv OPTIONS (path "obs://dli-test-021/data4") PARTITIONED BY (classNo,dt); 给 testcsvdata5source表插入如下测试数据： insert into testcsvdata5source VALUES('Aarn','98','25','2021-07-27'); insert into testcsvdata5source VALUES('Adam','68','25','2021-07-28'); 根据classNo分区列查询testcsvdata5source数据。 select * from testcsvdata5source where classNo = 25; 图12 查询结果 根据dt分区列查询testcsvdata5source数据。 select * from testcsvdata5source where dt like '2021-07%'; 图13 查询结果 在OBS桶“obs://dli-test-021/data4”目录下刷新后查询，会生成如下数据文件： 文件目录1：obs://dli-test-021/data4/xxxxxx/classNo=25/dt=2021-07-27 图14 查询结果 文件目录2：obs://dli-test-021/data4/xxxxxx/classNo=25/dt=2021-07-28 图15 查询结果 在OBS桶的“obs://dli-test-021/data4”目录下创建“classNo=24”的分区目录，再在“classNo=24”目录下创建子分区目录“dt=2021-07-29”。根据以下文件内容创建数据文件“test.csv”，并上传到OBS的“obs://dli-test-021/data4/classNo=24/dt=2021-07-29”目录下。 Jordon,88,24,2021-07-29 Kim,87,24,2021-07-29 Henry,76,24,2021-07-29 在SQL编辑器中执行以下命令，导入分区数据到OBS表“testcsvdata5source ”。 ALTER TABLE testcsvdata5source ADD PARTITION (classNo = 24,dt='2021-07-29') LOCATION 'obs://dli-test-021/data4/classNo=24/dt=2021-07-29'; 根据classNo分区列查询testcsvdata5source数据。 select * from testcsvdata5source where classNo = 24; 图16 查询结果 根据dt分区列查询所有“2021-07”月的所有数据。 select * from testcsvdata5source where dt like '2021-07%'; 图17 查询结果

流生态开发支持的数据格式 DLI Flink作业支持如下数据格式： Avro，Avro_merge，BLOB，CSV，EMAIL，JSON，ORC，Parquet，XML。 表4 数据格式和支持的输入输出流 数据格式 支持的输入流 支持的输出流 Avro - OBS输出流 Avro_merge - OBS输出流 BLOB DIS输入流 MRS Kafka输入流 开源Kafka输入流 - CSV DIS输入流 OBS输入流 开源Kafka输入流 DIS输出流 OBS输出流 DWS输出流（通过OBS方式转储） 开源Kafka输出流 文件系统输出流 EMAIL DIS输入流 - JSON DIS输入流 OBS输入流 MRS Kafka输入流 开源Kafka输入流 DIS输出流 OBS输出流 MRS Kafka输出流 开源Kafka输出流 ORC - OBS输出流 DWS输出流（通过OBS方式转储） Parquet - OBS输出流 文件系统输出流 XML DIS输入流 -

概述 流生态系统基于Flink和Spark双引擎，完全兼容Flink/Storm/Spark开源社区版本接口，并且在此基础上做了特性增强和性能提升，为用户提供易用、低时延、高吞吐的 数据湖探索 。 数据湖 探索的流生态开发包括云服务生态、开源生态和自拓展生态： 云服务生态 DLI服务在Stream SQL中支持与其他服务的连通。用户可以直接使用SQL从这些服务中读写数据，如DIS、OBS、CloudTable、MRS、RDS、 SMN 、DCS等。 开源生态 通过对等连接建立与其他VPC的网络连接后，用户可以在DLI的租户独享集群中访问所有Flink和Spark支持的数据源与输出源，如Kafka、Hbase、ElasticSearch等。 自拓展生态 用户可通过编写代码实现从想要的云生态或者开源生态获取数据，作为Flink作业的输入数据。

云服务生态开发 表1 云服务生态开发一览表 数据源 SQL 自定义作业 输入流：从其他服务或数据库中获取数据 输出流：将处理后的数据写入到其他服务或数据库中 表格存储服务 CloudTable HBase输入流 HBase输出流 OpenTSDB输出流 - 云搜索服务 CSS - Elasticsearch输出流 - 分布式缓存服务 DCS - DCS输出流 自定义作业交互 文档数据库服务 DDS - DDS输出流 - 数据接入服务 DIS DIS输入流 DIS输出流 - 分布式消息服务 DMS DMS输入流 DMS输出流 - 数据仓库服务 DWS - DWS输出流（通过JDBC方式转储） DWS输出流（通过OBS方式转储） 自定义作业交互 MapReduce服务 MRS MRS Kafka输入流 MRS Kafka输出流 MRS HBase输出流 自定义作业交互 对象存储服务 OBS OBS输入流 OBS输出流 - 关系型数据库 RDS - RDS输出流 - 消息通知 服务 SMN - SMN输出流 -

步骤3：创建DWS数据库和表 连接已创建的DWS集群。 请参考使用gsql命令行客户端连接DWS集群。 执行以下命令连接DWS集群的默认数据库“gaussdb”： gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r gaussdb：DWS集群默认数据库。 DWS集群连接地址：请参见获取集群连接地址进行获取。如果通过公网地址连接，请指定为集群“公网访问地址”或“公网访问 域名 ”，如果通过内网地址连接，请指定为集群“内网访问地址”或“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。 dbadmin：创建集群时设置的默认管理员用户名。 -W：默认管理员用户的密码。 在命令行窗口输入以下命令创建数据库“testdwsdb”。 CREATE DATABASE testdwsdb; 执行以下命令，退出gaussdb数据库，连接新创建的数据库“testdwsdb”。 \q gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r 执行以下命令创建表。 create schema test; set current_schema= test; drop table if exists dwsresult; CREATE TABLE dwsresult ( car_id VARCHAR, car_owner VARCHAR, car_age INTEGER , average_speed FLOAT8, total_miles FLOAT8 );

整体作业开发流程 整体作业开发流程参考图1。 图1 作业开发流程 步骤1：创建队列：创建DLI作业运行的队列。 步骤2：创建RDS MySQL数据库和表：创建RDS MySQL的数据库和表。 步骤3：创建DWS数据库和表：创建用于接收数据的DWS数据库和表。 步骤4：创建增强型跨源连接：DLI上创建连接RDS和DWS的跨源连接，打通网络。 步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。 步骤6：发送数据和查询结果：RDS MySQL的表上插入数据，在DWS上查看运行结果。

开发流程 DLI进行Spark作业访问DLI元数据开发流程参考如下： 图1 Spark作业访问DLI元数据开发流程 表2 开发流程说明 序号 阶段 操作界面 说明 1 创建DLI通用队列 DLI控制台 创建作业运行的DLI队列。 2 OBS桶文件配置 OBS控制台 如果是创建OBS表，则需要上传文件数据到OBS桶下。 配置Spark创建表的元数据信息的存储路径。该文件夹路径用来存储Spark创建表的元数据信息“spark.sql.warehouse.dir”。 3 新建Maven工程，配置pom文件 IntelliJ IDEA 参考样例代码说明，编写程序代码创建DLI表或OBS表。 4 编写程序代码 5 调试，编译代码并导出Jar包 6 上传Jar包到OBS和DLI OBS控制台 DLI控制台 将生成的Spark Jar包文件上传到OBS目录下和DLI程序包中。 7 创建Spark Jar作业 DLI控制台 在DLI控制台创建Spark Jar作业并提交运行作业。 8 查看作业运行结果 DLI控制台 查看作业运行状态和作业运行日志。

步骤4：编写代码 编写DliCatalogTest程序创建数据库、DLI表和OBS表。 完整的样例请参考Java样例代码，样例代码分段说明如下： 导入依赖的包。 import org.apache.spark.sql.SparkSession; 创建SparkSession会话。 创建SparkSession会话时需要指定Spark参数："spark.sql.session.state.builder"、"spark.sql.catalog.class"和"spark.sql.extensions"，按照样例配置即可。 Spark2.3.x版本 SparkSession spark = SparkSession .builder() .config("spark.sql.session.state.builder", "org.apache.spark.sql.hive.UQueryHiveACLSessionStateBuilder") .config("spark.sql.catalog.class", "org.apache.spark.sql.hive.UQueryHiveACLExternalCatalog") .config("spark.sql.extensions","org.apache.spark.sql.DliSparkExtension") .appName("java_spark_demo") .getOrCreate(); Spark2.4.x版本 SparkSession spark = SparkSession .builder() .config("spark.sql.session.state.builder", "org.apache.spark.sql.hive.UQueryHiveACLSessionStateBuilder") .config("spark.sql.catalog.class", "org.apache.spark.sql.hive.UQueryHiveACLExternalCatalog") .config("spark.sql.extensions","org.apache.spark.sql.DliSparkExtension") .config("spark.sql.hive.implementation","org.apache.spark.sql.hive.client.DliHiveClientImpl") .appName("java_spark_demo") .getOrCreate(); Spark3.1.x版本 SparkSession spark = SparkSession .builder() .config("spark.sql.session.state.builder", "org.apache.spark.sql.hive.UQueryHiveACLSessionStateBuilder") .config("spark.sql.catalog.class", "org.apache.spark.sql.hive.UQueryHiveACLExternalCatalog") .config("spark.sql.extensions","org.apache.spark.sql.DliSparkExtension") .appName("java_spark_demo") .getOrCreate(); Spark3.3.x版本 SparkSession spark = SparkSession .builder() .config("spark.sql.session.state.builder", "org.apache.spark.sql.hive.DliLakeHouseBuilder") .config("spark.sql.catalog.class", "org.apache.spark.sql.hive.DliLakeHouseCatalog") .appName("java_spark_demo") .getOrCreate(); 创建数据库。 如下样例代码演示，创建名为test_sparkapp的数据库。 spark.sql("create database if not exists test_sparkapp").collect(); 创建DLI表并插入测试数据。 spark.sql("drop table if exists test_sparkapp.dli_testtable").collect(); spark.sql("create table test_sparkapp.dli_testtable(id INT, name STRING)").collect(); spark.sql("insert into test_sparkapp.dli_testtable VALUES (123,'jason')").collect(); spark.sql("insert into test_sparkapp.dli_testtable VALUES (456,'merry')").collect(); 创建OBS表。如下示例中的OBS路径需要根据步骤2：OBS桶文件配置中的实际数据路径修改。 spark.sql("drop table if exists test_sparkapp.dli_testobstable").collect(); spark.sql("create table test_sparkapp.dli_testobstable(age INT, name STRING) using csv options (path 'obs://dli-test-obs01/testdata.csv')").collect(); 关闭SparkSession会话spark。 spark.stop();