华为云用户手册

语法格式 在Spark Jar作业编辑界面，选择配置优化参数，配置信息如下: 不同的OBS桶，使用不同的AKSK认证信息。 可以使用如下配置方式，根据桶指定不同的AKSK信息，参数说明详见表1。 spark.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key= USER_AK_ CS MS_KEY spark.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key= USER_SK_C SMS _KEY spark.hadoop.fs.obs.security.provider = com.dli.provider.UserObsBasicCredentialProvider spark.hadoop.fs.dew.csms.secretName= CredentialName spark.hadoop.fs.dew.endpoint=ENDPOINT spark.hadoop.fs.dew.csms.version=VERSION_ID spark.hadoop.fs.dew.csms.cache.time.second =CACHE_TIME spark.dli.job.agency.name=USER_AGENCY_NAME

前提条件 已在DEW服务创建通用凭证，并存入凭据值。具体操作请参考：创建通用凭据。 已创建 DLI 访问DEW的委托并完成委托授权。该委托需具备以下权限： DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。 DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。 DEW解密凭据的权限，kms:dek:decrypt。 委托权限示例请参考自定义DLI委托权限和常见场景的委托权限策略。 仅支持Flink1.15版本使用DEW管理访问凭据，在创建作业时，请配置作业使用Flink1.15版本、且已在作业中配置允许DLI访问DEW的委托信息。自定义委托及配置请参考自定义DLI委托权限。 使用该功能，所有涉及OBS的桶，都需要进行配置AKSK。

操作场景 DLI将Flink Jar作业的输出数据写入到OBS时，需要配置AKSK访问OBS，为了确保AKSK数据安全，您可以用过 数据加密 服务（Data Encryption Workshop，DEW）、云凭据管理服务（Cloud Secret Management Service，CSMS），对AKSK统一管理，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。 本例以获取访问OBS的AKSK为例介绍Flink Jar使用DEW获取访问凭证读写OBS的操作指导。

参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 参数说明 flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key 是 无 String USER_BUCKET_NAME为用户的桶名，需要进行替换为用户的使用的OBS桶名。 参数的值为用户定义在CSMS通用凭证中的键key， 其Key对应的value为用户的AK（Access Key Id），需要具备访问OBS对应桶的权限。 flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key 是 无 String USER_BUCKET_NAME为用户的桶名，需要进行替换为用户的使用的OBS桶名。 参数的值为用户定义在CSMS通用凭证中的键key， 其Key对应的value为用户的SK（Secret Access Key），需要具备访问OBS对应桶的权限。 flink.hadoop.fs.obs.security.provider 是 无 String OBS AKSK认证机制，使用DEW服务中的CSMS凭证管理，获取OBS的AK、SK。 默认取值为com.dli.provider.UserObsBasicCredentialProvider flink.hadoop.fs.dew.endpoint 是 无 String 指定要使用的DEW服务所在的endpoint信息。 获取地区和终端节点。 配置示例：flink.hadoop.fs.dew.endpoint=kms.cn-xxxx.myhuaweicloud.com flink.hadoop.fs.dew.projectId 否 有 String DEW所在的项目ID， 默认是Flink作业所在的项目ID。 获取项目ID flink.hadoop.fs.dew.csms.secretName 是 无 String 在DEW服务的凭据管理中新建的通用凭据的名称。 配置示例：flink.hadoop.fs.dew.csms.secretName=secretInfo flink.hadoop.fs.dew.csms.version 否 最新的version String 在DEW服务的凭据管理中新建的通用凭据的版本号(凭据的版本标识符)。 若不指定，则默认获取该通用凭证的最新版本号。 配置示例：flink.hadoop.fs.dew.csms.version=v1 flink.hadoop.fs.dew.csms.cache.time.second 否 3600 Long Flink作业访问获取CSMS通用凭证后，缓存的时间。 单位为秒。默认值为3600秒。 flink.dli.job.agency.name 是 - String 自定义委托名称。

语法格式 在Flink jar作业编辑界面，选择配置优化参数，配置信息如下: 不同的OBS桶，使用不同的AKSK认证信息。 可以使用如下配置方式，根据桶指定不同的AKSK信息，参数说明详见表1。 flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key=USER_AK_CSMS_KEY flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key=USER_SK_CSMS_KEY flink.hadoop.fs.obs.security.provider=com.dli.provider.UserObsBasicCredentialProvider flink.hadoop.fs.dew.csms.secretName=CredentialName flink.hadoop.fs.dew.endpoint=ENDPOINT flink.hadoop.fs.dew.csms.version=VERSION_ID flink.hadoop.fs.dew.csms.cache.time.second=CACHE_TIME flink.dli.job.agency.name=USER_AGENCY_NAME

示例 本例以通过DataGen表产生随机数据并输出到Mysql结果表中为例，介绍Flink Opensource SQL使用DEW管理访问凭据的配置方法。 创建DLI与Mysql的增强型跨源连接。详细步骤请参考创建增强型跨源连接。 创建DLI访问DEW的委托并完成委托授权。详细步骤请参考自定义DLI委托权限。 在DEW创建通用凭证。详细操作请参考创建通用凭据。 登录DEW管理控制台 选择“凭据管理”，进入“凭据管理”页面。 单击“创建凭据”。配置凭据基本信息 本例配置Mysql的凭据值： "MySQLUsername":"demo" "MySQLPassword":"*******",其中"*******"为Mysql的访问密码 Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。 create table dataGenSource( user_id string, amount int ) with ( 'connector' = 'datagen', 'rows-per-second' = '1', --每秒生成一条数据 'fields.user_id.kind' = 'random', --为字段user_id指定random生成器 'fields.user_id.length' = '3' --限制user_id长度为3 ); CREATE TABLE jdbcSink ( user_id string, amount int ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--其中url中的flink表示MySQL中orders表所在的数据库名 'table-name' = 'orders', 'username' = 'MySQLUsername', -- DEW服务中，名称为secretInfo，且版本号v1的的通用凭证，定义凭证值的键MySQLUsername，它的值为用户的敏感信息。 'password' = 'MySQLPassword', -- DEW服务中，名称为secretInfo，且版本号v1的的通用凭证，定义凭证值的键MySQLPassword，它的值为用户的敏感信息。 'sink.buffer-flush.max-rows' = '1', 'dew.endpoint'='kms.cn-xxxx.myhuaweicloud.com', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName'='secretInfo', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields'='username,password', --其中username,password字段值，需要利用DEW凭证管理,进行解密替换。 'dew.csms.version'='v1' ); insert into jdbcSink select * from dataGenSOurce;

语法格式 create table tableName( attr_name attr_type (',' attr_name attr_type)* (',' WATERMARK FOR rowtime_column_name AS watermark-strategy_expression) ) with ( ... 'dew.endpoint'='', 'dew.csms.secretName'='', 'dew.csms.decrypt.fields'='', 'dew.projectId'='', 'dew.csms.version'='' );

参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 参数说明 dew.endpoint 是 无 String 指定要使用的DEW服务所在的endpoint信息。 获取地区和终端节点。 配置示例：'dew.endpoint'='kms.cn-xxxx.myhuaweicloud.com' dew.projectId 否 有 String DEW所在的项目ID， 默认是Flink作业所在的项目ID。 获取项目ID dew.csms.secretName 是 无 String 在DEW服务的凭据管理中新建的通用凭据的名称。 配置示例：'dew.csms.secretName'='secretInfo' dew.csms.decrypt.fields 是 无 String 指定connector with属性中，哪些字段属性需要使用DEW云凭据管理服务进行解密。字段属性之间用逗号分隔，例如：'dew.csms.decrypt.fields'='field1,field2,field3' dew.csms.version 否 最新的version String 在DEW服务的凭据管理中新建的通用凭据的版本号(凭据的版本标识符)。 若不指定，则默认获取该通用凭证的最新版本号。 配置示例： 'dew.csms.version'='v1'

前提条件 已在DEW服务创建通用凭证，并存入凭据值。具体操作请参考：创建通用凭据。 已创建DLI访问DEW的委托并完成委托授权。该委托需具备以下权限： DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。 DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。 DEW解密凭据的权限，kms:dek:decrypt。 委托权限示例请参考自定义DLI委托权限和常见场景的委托权限策略。 仅支持Flink1.15版本使用DEW管理访问凭据，在创建作业时，请配置作业使用Flink1.15版本、且已在作业中配置允许DLI访问DEW的委托信息。 在DLI管理控制台新建“增强型跨源连接”配置DLI与数据源的网络连通。 具体操作请参考增强型跨源连接。

操作场景 DLI将Flink作业的输出数据写入到Mysql或DWS时，需要在connector中设置账号、密码等属性。但是账号密码等信息属于高度敏感数据，需要做加密处理，以保障用户的数据隐私安全。 数据加密服务（Data Encryption Workshop，DEW）、云凭据管理服务（Cloud Secret Management Service，CSMS），提供一种安全、可靠、简单易用隐私数据加解密方案。 用户或应用程序通过凭据管理服务，创建、检索、更新、删除凭据，轻松实现对敏感凭据的全生命周期的统一管理，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。 本节操作介绍Flink Opensource SQL场景使用DEW管理和访问凭据的操作指导。

开发流程 DLI下UDTF函数开发流程参考如下： 图1 UDTF开发流程 表2 开发流程说明 序号 阶段 操作界面 说明 1 新建Maven工程，配置pom文件 IntelliJ IDEA 参考操作步骤说明，编写UDTF函数代码。 2 编写UDTF函数代码 3 调试，编译代码并导出Jar包 4 上传Jar包到OBS OBS控制台 将生成的UDTF函数Jar包文件上传到OBS目录下。 5 创建DLI的UDTF函数 DLI控制台 在DLI控制台的SQL作业管理界面创建使用的UDTF函数。 6 验证和使用DLI的UDTF函数 DLI控制台 在DLI作业中使用创建的UDTF函数。

通过DLI Livy工具提交Spark作业到DLI 本示例演示通过curl命令使用DLI Livy工具将Spark作业提交到DLI。 将开发好的Spark作业程序jar包上传到OBS路径下。 例如，本示例上传“spark-examples_2.11-XXXX.jar”到“obs://bucket/path”路径下。 以root用户登录到安装DLI Livy工具的ECS服务器。 执行curl命令通过DLI Livy工具提交Spark作业请求到DLI。 ECS_IP为当前安装DLI Livy工具所在的弹性云服务器的私有IP地址。 curl --location --request POST 'http://ECS_IP:8998/batches' \ --header 'Content-Type: application/json' \ --data '{ "driverMemory": "3G", "driverCores": 1, "executorMemory": "2G", "executorCores": 1, "numExecutors": 1, "args": [ "1000" ], "file": "obs://bucket/path/spark-examples_2.11-XXXX.jar", "className": "org.apache.spark.examples.SparkPi", "conf": { "spark.dynamicAllocation.minExecutors": 1, "spark.executor.instances": 1, "spark.dynamicAllocation.initialExecutors": 1, "spark.dynamicAllocation.maxExecutors": 2 } }'

修改DLI Livy工具配置文件 上传指定的DLI Livy工具jar资源包到OBS桶路径下。 登录OBS控制台，在指定的OBS桶下创建一个存放Livy工具jar包的资源目录。例如：“obs://bucket/livy/jars/”。 进入3.a中DLI Livy工具所在ECS服务器的安装目录，获取以下jar包，将获取的jar包上传到1.a创建的OBS桶资源目录下。 例如，当前Livy工具安装路径为“/opt/livy”，则当前需要上传的jar包名称如下： /opt/livy/rsc-jars/livy-api-0.7.2.0107.jar /opt/livy/rsc-jars/livy-rsc-0.7.2.0107.jar /opt/livy/repl_2.11-jars/livy-core_2.11-0.7.2.0107.jar /opt/livy/repl_2.11-jars/livy-repl_2.11-0.7.2.0107.jar 修改DLI Livy工具配置文件。 编辑修改配置文件“ /opt/livy/conf/livy-client.conf”。 vi /opt/livy/conf/livy-client.conf 添加如下内容，并根据注释修改配置项。 #当前ECS的私有IP地址，也可以使用ifconfig命令查询。 livy.rsc.launcher.address = X.X.X.X #当前ECS服务器放通的端口号 livy.rsc.launcher.port.range = 30000~32767 编辑修改配置文件“ /opt/livy/conf/livy.conf”。 vi /opt/livy/conf/livy.conf 添加如下内容。根据注释说明修改具体的配置项。 livy.server.port = 8998 livy.spark.master = yarn livy.server.contextLauncher.custom.class=org.apache.livy.rsc.DliContextLauncher livy.server.batch.custom.class=org.apache.livy.server.batch.DliBatchSession livy.server.interactive.custom.class=org.apache.livy.server.interactive.DliInteractiveSession livy.server.sparkApp.custom.class=org.apache.livy.utils.SparkDliApp livy.server.recovery.mode = recovery livy.server.recovery.state-store = filesystem #以下文件路径请根据情况修改 livy.server.recovery.state-store.url = file:///opt/livy/store/ livy.server.session.timeout-check = true livy.server.session.timeout = 1800s livy.server.session.state-retain.sec = 1800s livy.dli.spark.version = 2.3.2 livy.dli.spark.scala-version = 2.11 # 填入存储livy jar包资源的OBS桶路径。 livy.repl.jars = obs://bucket/livy/jars/livy-core_2.11-0.7.2.0107.jar, obs://bucket/livy/jars/livy-repl_2.11-0.7.2.0107.jar livy.rsc.jars = obs://bucket/livy/jars/livy-api-0.7.2.0107.jar, obs://bucket/livy/jars/livy-rsc-0.7.2.0107.jar 编辑修改配置文件“/opt/livy/conf/spark-defaults.conf”。 vi /opt/livy/conf/spark-defaults.conf 添加如下必选参数内容。配置项参数填写说明，详见表1。 # 以下参数均支持在提交作业时覆盖。 spark.yarn.isPython=true spark.pyspark.python=python3 # 当前参数值为生产环境web地址 spark.dli.user.uiBaseAddress=https://console.huaweicloud.com/dli/web # 队列所在的region。 spark.dli.user.regionName=XXXX # dli endpoint 地址。 spark.dli.user.dliEndPoint=XXXX # 用于指定队列，填写已创建DLI的队列名。 spark.dli.user.queueName=XXXX # 提交作业使用的access key。 spark.dli.user.access.key=XXXX # 提交作业使用的secret key。 spark.dli.user.secret.key=XXXX # 提交作业使用的projectId。 spark.dli.user.projectId=XXXX 表1 spark-defaults.conf必选参数说明 参数名 参数填写说明 spark.dli.user.regionName DLI队列所在的区 域名 。 从地区和终端节点获取，对应“区域”列就是regionName。 spark.dli.user.dliEndPoint DLI队列所在的终端节点。 从地区和终端节点获取，对应的“终端节点（Endpoint）”就是该参数取值。 spark.dli.user.queueName DLI队列名称。 spark.dli.user.access.key 对应用户的访问密钥。该用户需要有Spark作业相关权限，权限说明详见权限管理。 密钥获取方式请参考获取AK/SK。 spark.dli.user.secret.key spark.dli.user.projectId 参考获取项目ID获取项目ID。 以下参数为可选参数，请根据参数说明和实际情况配置。详细参数说明请参考Spark Configuration。 表2 spark-defaults.conf可选参数说明 Spark作业参数 对应Spark批处理参数 备注 spark.dli.user.file file 如果是对接notebook工具场景时不需要设置。 spark.dli.user.className class_name 如果是对接notebook工具场景时不需要设置。 spark.dli.user.scType sc_type 推荐使用livy原生配置。 spark.dli.user.args args 推荐使用livy原生配置。 spark.submit.pyFiles python_files 推荐使用livy原生配置。 spark.files files 推荐使用livy原生配置。 spark.dli.user.modules modules - spark.dli.user.image image 提交作业使用的 自定义镜像 ，仅容器集群支持该参数，默认不设置。 spark.dli.user.autoRecovery auto_recovery - spark.dli.user.maxRetryTimes max_retry_times - spark.dli.user.catalogName catalog_name 访问元数据时，需要将该参数配置为dli。

准备工作 创建DLI队列。在“队列类型”中选择“通用队列”，即Spark作业的计算资源。具体请参考创建队列。 准备一个linux弹性 云服务器ECS ，用于安装DLI Livy。 ECS需要放通30000至32767端口、8998端口。具体操作请参考添加安全组规则。 ECS需安装Java JDK，JDK版本建议为1.8。配置Java环境变量JAVA_HOME。 查询弹性云服务器ECS详细信息，获取ECS的“私有IP地址”。 使用增强型跨源连接打通DLI队列和Livy实例所在的VPC网络。具体操作可以参考增强型跨源连接。

DLI Livy工具下载及安装 本次操作下载的DLI Livy版本为apache-livy-0.7.2.0107-bin.tar.gz，后续版本变化请根据实际情况修改。 单击下载链接，获取DLI Livy工具压缩包。 使用WinSCP工具，将获取的工具压缩包上传到准备好的ECS服务器目录下。 使用root用户登录ECS服务器，执行以下命令安装DLI Livy工具。 执行以下命令创建工具安装路径。 mkdir livy安装路径 例如新建路径/opt/livy：mkdir /opt/livy。后续操作步骤均默认以/opt/livy安装路径演示，请根据实际情况修改。 解压工具压缩包到安装路径。 tar --extract --file apache-livy-0.7.2.0107-bin.tar.gz --directory /opt/livy --strip-components 1 --no-same-owner 执行以下命令修改配置文件名称。 cd /opt/livy/conf mv livy-client.conf.template livy-client.conf mv livy.conf.template livy.conf mv livy-env.sh.template livy-env.sh mv log4j.properties.template log4j.properties mv spark-blacklist.conf.template spark-blacklist.conf touch spark-defaults.conf

问题1：运行Spark作业，作业运行失败，作业日志中提示java server connection或container启动失败 问题现象 运行Spark作业，作业运行失败，作业日志中提示java server connection或container启动失败。 解决方案 确认是否已修改跨源连接的主机信息，如果没有，请参考DLI跨源连接中配置 MRS 主机信息修改主机信息。重新创建和提交Spark作业。

问题2：运行Spark作业，作业运行失败，作业日志中提示KrbException: Message stream modified (41) 问题现象 运行Spark作业，作业运行失败，作业日志中提示KrbException: Message stream modified (41) 解决方案 编辑“krb5.conf” 配置文件，将文件中所有“renew_lifetime = xxx”配置删除。重新创建和提交Spark作业。

创建MRS HBase表 创建DLI表关联MRS HBase表之前确保HBase的表是存在的。以样例代码为例，具体的流程是： 远程登录ECS，通过hbase shell命令查看表信息。其中，“hbtest”是要查询的表名。 describe 'hbtest' （可选）如果不存在对应的HBase表，可以创建该表，具体的命令是： create ‘hbtest’, ‘info’, ‘detail’ 其中，“hbtest”是表名，其余为列族名。 配置好连接信息。“TableName”对应HBase表的表名，“Rowkey”和“Cols”请参考《创建DLI表关联HBase》进行配置。

CSS 安全集群参数配置 具体参数请参考表1，这里主要说明配置CSS安全集群连接参数时需要注意的内容。 .option("es.net.http.auth.user", "admin") .option("es.net.http.auth.pass", "***") 此处的参数为身份验证的账号和密码，也是登录Kibana的账号和密码。 .option("es.net.ssl", "true") 如果CSS安全集群开启了HTTPS访问，此处需要设置为“true”，并且需要继续设置后面的安全证书、文件地址等参数。 如果CSS安全集群未开启HTTPS访问，此处需要设置为“false”，则不需要设置后面安全证书、文件地址等参数。 .option("es.net.ssl.keystore.location", "obs://桶名/path/transport-keystore.jks") .option("es.net.ssl.keystore.pass", "***") 此处设置keystore.jks文件的位置以及进入这个文件的密钥。在准备工作中生成的keystore.jks文件需要先放到OBS桶中，然后填入ak和sk以及jks文件的具体位置。最后在“es.net.ssl.keystore.pass”填入进入文件的密钥。 .option("es.net.ssl.truststore.location", "obs://桶名/path/truststore.jks") .option("es.net.ssl.truststore.pass", "***") 此处是truststore.jks文件的设置参数，与keystore.jks文件的设置参数基本一致，按照keystore.jks文件设置步骤进行操作即可。

准备工作 当前CSS服务提供的Elasticsearch 6.5.4或以上集群版本为用户增加了安全模式功能，开启安全模式后，将会为用户提供身份验证、授权以及加密等功能。DLI服务对接CSS安全集群时，需要先进行以下准备工作。 选择CSS Elasticsearch 6.5.4或以上集群版本，创建CSS安全集群，并下载安全集群证书（CloudSearchService.cer）。 登录 云搜索服务 控制台，单击“集群管理”，选择需要建立跨源连接的集群，如图1所示： 图1 CSS集群管理 单击“安全模式”中的“下载证书”下载安全证书。 使用keytool工具生成keystore和truststore文件。 使用keytool工具生成keystore和truststore文件，其中需要使用到安全集群的安全证书（CloudSearchService.cer），keytool工具还有其他参数，可根据需求设置。 打开cmd，输入下列命令生成含有一个私钥的keystore文件。 keytool -genkeypair -alias certificatekey -keyalg RSA -keystore transport-keystore.jks 使用keytool工具生成keystore和truststore文件后，可以在文件夹中看到transport-keystore.jks文件，使用如下命令验证keystore文件和证书信息。 keytool -list -v -keystore transport-keystore.jks 正确输入keystore password后即可看见相应的信息。 使用如下命令创建truststore.jks文件并进行验证 。 keytool -import -alias certificatekey -file CloudSearchService.cer -keystore truststore.jks keytool -list -v -keystore truststore.jks 将生成的keystore和truststore文件上传到OBS桶中。

开发流程 DLI下UDF函数开发流程参考如下： 图1 开发流程 表2 开发流程说明 序号 阶段 操作界面 说明 1 新建Maven工程，配置pom文件 IntelliJ IDEA 参考操作步骤说明，编写UDF函数代码。 2 编写UDF函数代码 3 调试，编译代码并导出Jar包 4 上传Jar包到OBS OBS控制台 将生成的UDF函数Jar包文件上传到OBS目录下。 5 创建DLI的UDF函数 DLI控制台 在DLI控制台的SQL作业管理界面创建使用的UDF函数。 6 验证和使用DLI的UDF函数 DLI控制台 在DLI作业中使用创建的UDF函数。

流生态开发支持的数据格式 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输出流 -

支持的API列表 DLI JDBC Driver支持的API列表如下，对可能与JDBC标准产生歧义的地方加以备注说明。 Connection API支持的常用方法签名： Statement createStatement() PreparedStatement prepareStatement(String sql) void close() boolean isClosed() DatabaseMetaData getMetaData() PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) Driver API支持的常用方法签名： Connection connect(String url, Properties info) boolean acceptsURL(String url) DriverPropertyInfo[] getPropertyInfo(String url, Properties info) ResultSetMetaData API支持的常用方法签名： String getColumnClassName(int column) int getColumnCount() int getColumnDisplaySize(int column) String getColumnLabel(int column) String getColumnName(int column) int getColumnType(int column) String getColumnTypeName(int column) int getPrecision(int column) int getScale(int column) boolean isCaseSensitive(int column) Statement API支持的常用方法签名： ResultSet executeQuery(String sql) int executeUpdate(String sql) boolean execute(String sql) void close() int getMaxRows() void setMaxRows(int max) int getQueryTimeout() void setQueryTimeout(int seconds) void cancel() ResultSet getResultSet() int getUpdateCount() boolean isClosed() PreparedStatement API支持的常用方法签名： void clearParameters() boolean execute() ResultSet executeQuery() int executeUpdate() PreparedStatement Set系列方法 ResultSet API支持的常用方法签名： int getRow() boolean isClosed() boolean next() void close() int findColumn(String columnLabel) boolean wasNull() get系列方法 DatabaseMetaData API支持的常用方法签名 ResultSet getCatalogs() 在DLI服务中没有Catalog的概念，返回空的ResultSet。 ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern) Connection getConnection() getTables(String catalog, String schemaPattern,String tableNamePattern, String types[]) 该方法不采纳Catalog参数，schemaPattern对应DLI服务的database的概念。 ResultSet getTableTypes() ResultSet getSchemas() ResultSet getSchemas(String catalog, String schemaPattern)

概述 DLI支持原生Spark的DataSource能力，并在其基础上进行了扩展，能够通过SQL语句或者Spark作业访问其他数据存储服务并导入、查询、分析处理其中的数据，目前支持的DLI跨源访问服务有： 表格存储服务 CloudTable，云搜索服务CSS，分布式缓存服务DCS，文档数据库服务DDS， 数据仓库 服务 GaussDB （DWS），MapReduce服务MRS，云数据库RDS等。使用DLI的跨源能力，需要先创建跨源连接。 管理控制台界面具体操作请参考《数据湖探索用户指南》。 使用Spark作业跨源访问数据源支持使用scala，pyspark和java三种语言进行开发。 表格存储 服务CloudTable相关介绍请参考《表格存储服务产品介绍》。 云搜索服务CSS相关介绍请参考《云搜索服务产品介绍》。 分布式缓存服务DCS相关介绍请参考《分布式缓存服务产品介绍》。 文档数据库服务DDS相关介绍请参考《文档数据库服务产品介绍》。 数据仓库服务GaussDB（DWS）相关介绍请参考《数据仓库服务产品介绍》。 MapReduce服务MRS相关介绍请参考《MapReduce服务产品介绍》。 云数据库RDS相关介绍请参考《云数据库服务产品介绍》。 父主题： 使用Spark作业跨源访问数据源

开发流程 DLI下UDAF函数开发流程参考如下： 图1 UDAF开发流程 表2 开发流程说明 序号 阶段 操作界面 说明 1 新建Maven工程，配置pom文件 IntelliJ IDEA 参考操作步骤说明，编写UDAF函数代码。 2 编写UDAF函数代码 3 调试，编译代码并导出Jar包 4 上传Jar包到OBS OBS控制台 将生成的UDAF函数Jar包文件上传到OBS目录下。 5 创建DLI程序包 DLI控制台 选择刚上传到OBS的UDAF函数的Jar文件，由DLI进行纳管。 6 创建DLI的UDAF函数 DLI控制台 在DLI控制台的SQL作业管理界面创建使用的UDAF函数。 7 验证和使用DLI的UDAF函数 DLI控制台 在DLI作业中使用创建的UDAF函数。

环境准备 在进行UDAF开发前，请准备以下开发环境。 表1 UDAF开发环境 准备项 说明 操作系统 Windows系统，支持Windows7以上版本。 安装JDK JDK使用1.8版本（访问Java官网）。 安装和配置IntelliJ IDEA IntelliJ IDEA为进行应用开发的工具，版本要求使用2019.1或其2019.1往后的版本。 安装Maven 开发环境的基本配置（下载并安装 Maven）。用于项目管理，贯穿软件开发生命周期。

常见问题 问题一：查询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表。