华为云用户手册

查询作业执行计划图 DLI 提供查询Flink作业执行计划图的接口。您可以使用该接口查询作业的执行计划图。示例代码如下： 1 2 3 4 5 private static void getFlinkJobExecuteGraph(DLIClient client) throws DLIException { Long jobId = 203L;//作业ID FlinkJobExecutePlanResponse result = client.getFlinkJobExecuteGraph(jobId); System.out.println(result); }

新建自定义作业 DLI提供新建Flink自定义作业的接口。您可以使用该接口创建一个用户自定义作业，目前支持jar格式，运行在独享队列中。示例代码如下： 1 2 3 4 5 6 7 8 9 10 private static void createFlinkJob(DLIClient client) throws DLIException { CreateFlinkJarJobRequest body = new CreateFlinkJarJobRequest(); body.name("jar-job"); body.cuNumber(2); body.managerCuNumber(1); body.parallelNumber(1); body.entrypoint("dli/WindowJoin.jar"); JobStatusResponse result = client.createFlinkJarJob(body); System.out.println(result); }

更新SQL作业 DLI提供更新Flink SQL作业接口。您可以使用该接口更新Flink SQL作业，示例代码如下： 1 2 3 4 5 6 private static void updateSQLJob(DLIClient client) throws DLIException { UpdateFlinkSqlJobRequest body = new UpdateFlinkSqlJobRequest(); body.name("update-job"); JobUpdateResponse result = client.updateFlinkSqlJob(body,203L); System.out.println(result); }

新建SQL作业 DLI提供新建Flink SQL作业的接口。您可以使用该接口新建Flink SQL作业并提交到DLI，示例代码如下： 1 2 3 4 5 6 7 8 9 10 private static void createSQLJob(DLIClient client) throws DLIException { SubmitFlinkSqlJobRequest body = new SubmitFlinkSqlJobRequest(); body.name("job-name"); body.runMode(SubmitFlinkSqlJobRequest.RunModeEnum.SHARED_CLUSTER); body.checkpointEnabled(false); body.checkpointMode(1); body.jobType(SubmitFlinkSqlJobRequest.JobTypeEnum.JOB); JobStatusResponse result = client.submitFlinkSqlJob(body); System.out.println(result); }

查询作业列表 DLI提供查询Flink作业列表的接口。您可以使用该接口查询作业列表。作业列表查询支持以下参数: name，status，show_detail，cursor，next，limit，order。本示例排序方式选择降序desc，将会列出作业id小于cursor的作业列表信息。示例代码如下： 1 2 3 4 private static void QueryFlinkJobListResponse(DLIClient client) throws DLIException { QueryFlinkJobListResponse result = client.getFlinkJobs(null, "job_init", null, true, 0L, 10, null, null,null,null,null); System.out.println(result); }

查询作业详情 DLI提供查询Flink作业详情的接口。您可以使用该接口查询作业的详情。示例代码如下： 1 2 3 4 5 private static void getFlinkJobDetail(DLIClient client) throws DLIException { Long jobId = 203L;//作业ID GetFlinkJobDetailResponse result = client.getFlinkJobDetail(jobId); System.out.println(result); }

更新自定义作业 DLI提供更新Flink自定义作业的接口。您可以使用该接口更新已经创建的自定义作业，目前仅支持Jar格式和运行在独享队列中。示例代码如下： 1 2 3 4 5 6 private static void updateFlinkJob(DLIClient client) throws DLIException { UpdateFlinkJarJobRequest body = new UpdateFlinkJarJobRequest(); body.name("update-job"); JobUpdateResponse result = client.updateFlinkJarJob(body,202L); System.out.println(result); }

使用须知 要使用DLI Python SDK访问指定服务的 API ，您需要确认已在DLI管理控制台开通当前服务并完成服务授权。 Python版本建议使用2.7.10和3.4.0以上版本，需要配置Visual C++编译环境Visual C++ build tools 或者 Visual Studio。 关于Python开发环境的配置请参考Python SDK环境配置。 DLI Python SDK依赖第三方库包括：urllib3 1.15以上版本，six 1.10以上版本，certifi，python-dateutil。 关于Python SDK的获取与安装请参考Python SDK获取与安装。 使用SDK工具访问DLI，需要用户初始化DLI客户端。用户可以使用AK/SK(Access Key ID/Secret Access Key)或Token两种认证方式初始化客户端，具体操作请参考初始化DLI客户端

Python SDK列表 表1 Python SDK列表 类型 说明 队列相关 介绍查询所有队列的Python SDK使用说明。 资源相关 介绍上传资源包、查询所有资源包、查询制定资源包、删除资源包的Python SDK使用说明。 SQL作业相关 介绍数据库相关、表相关、作业相关的Python SDK使用说明。 Spark作业相关 介绍提交Spark作业、取消Spark作业、删除Spark作业等Python SDK使用说明。

更新作业模板 DLI提供更新Flink作业模板的接口。您可以使用该接口修改一个Flink作业模板。示例代码如下： 1 2 3 4 5 6 7 public static void updateFlinkJobTemplate(DLIClient client) throws DLIException{ Long templateId = 277L;//模板Id UpdateFlinkJobTemplateRequest body = new UpdateFlinkJobTemplateRequest(); body.name("template-update"); GlobalResponse result = client.updateFlinkJobTemplate(body,templateId); System.out.println(result); }

删除作业模板 DLI提供删除Flink作业模板的接口。您可以使用该接口删除已经创建的作业模板，如果当前模板被引用也允许删除模板。示例代码如下： 1 2 3 4 5 public static void deleteFlinkJobTemplate(DLIClient client) throws DLIException{ Long templateId = 277L;//模板Id FlinkJobTemplateDeleteResponse result = client.deleteFlinkJobTemplate(templateId); System.out.println(result); }

新建作业模板 DLI提供新建Flink作业模板的接口。您可以使用该接口新建一个Flink作业模板。示例代码如下： 1 2 3 4 5 6 public static void createFlinkJobTemplate(DLIClient client) throws DLIException{ CreateFlinkJobTemplateRequest body = new CreateFlinkJobTemplateRequest(); body.name("template"); FlinkJobTemplateCreateResponse result = client.createFlinkJobTemplate(body); System.out.println(result); }

查询作业模板列表 DLI提供查询Flink作业模板的接口。您可以使用该接口查询作业模板列表。本示例排序方式选择降序desc，将会列出作业模板ID小于cursor的作业模板列表信息。示例代码如下： 1 2 3 4 5 6 7 public static void getFlinkJobTemplates(DLIClient client) throws DLIException{ Long offset = 789L; // Long | 模板偏移量。 Integer limit = 56; // Integer | 查询条数限制 String order = "asc"; // String | 查询结果排序, 升序和降序两种可选 FlinkJobTemplateListResponse result = client.getFlinkJobTemplates(offset,limit,order); System.out.println(result); }

SDK列表 表1提供了通用SDK V3的DLI支持列表，您可以在GitHub仓库查看SDK更新历史、获取安装包以及查看指导文档。 表1 SDK列表 编程语言 Github地址 参考文档 视频指导 Java huaweicloud-sdk-java-v3 Java SDK使用指导 Java SDK视频指导 Python huaweicloud-sdk-python-v3 Python SDK使用指导 Python SDK视频指导 PHP huaweicloud-sdk-php-v3 PHP SDK使用指导 PHP SDK视频指导 Go huaweicloud-sdk-go-v3 Go SDK使用指导 Go SDK视频指导 Node.js huaweicloud-sdk-nodejs-v3 Node.js SDK使用指导 Node.js SDK视频指导 .NET huaweicloud-sdk-net-v3 .NET SDK使用指导 .NET SDK视频指导 表1提供了DLI云服务支持的SDK V3列表，您可以在GitHub仓库查看SDK更新历史、获取安装包以及查看指导文档。

删除资源包 您可以使用该接口删除已上传的资源包，示例代码如下： def delete_resource(dli_client, resource_name, group_name): try: dli_client.delete_resource(resource_name, group_name) except DliException as e: print(e) return

上传资源包 您可以使用DLI提供的接口上传资源包，示例代码如下。完整样例代码和依赖包说明请参考：Python SDK概述。 1 2 3 4 5 6 def upload_resource(dli_client, kind, obs_jar_paths, group_name): try: dli_client.upload_resource(kind, obs_jar_paths, group_name) except DliException as e: print(e) return 请求参数说明如下，详细参数使用可以参考Python SDK概述下载样例代码。 kind：资源包类型，当前支持的包类型分别为： jar：用户jar文件 pyfile：用户Python文件 file：用户文件 modelfile：用户AI模型文件 obs_jar_paths：对应资源包的OBS路径，参数构成为：{bucketName}.{obs 域名 }/{jarPath}/{jarName}。 例如："https://bucketname.obs.cn-north-1.myhuaweicloud.com/jarname.jar" group_name：资源包所属分组名称。

查询所有资源包 DLI提供查询资源列表接口，您可以使用该接口并选择相应的资源来执行作业。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 def list_resources(dli_client): try: resources = dli_client.list_resources() except DliException as e: print(e) return for resources_info in resources.package_resources: print('Package resource name:' + resources_info.resource_name) for group_resource in resources.group_resources: print('Group resource name:' + group_resource.group_name) 完整样例代码和依赖包说明请参考：Python SDK概述。

查询指定资源包 您可以使用该接口查询指定的资源包信息，示例代码如下： def get_package_resource(dli_client, resource_name, group_name): try: pkg_resource = dli_client.get_package_resource(resource_name, group_name) print(pkg_resource) except DliException as e: print(e) return

AK/SK认证方式样例代码 代码样例 1 2 3 4 5 6 String ak = System.getenv("xxx_SDK_AK");//访问密钥ID。 String sk = System.getenv("xxx_SDK_SK");//与访问密钥ID结合使用的密钥。 String regionName = "regionname"; String projectId = "project_id"; DLIInfo dliInfo = new DLIInfo(regionName, ak, sk, projectId); DLIClient client = new DLIClient(AuthenticationMode.AKSK, dliInfo); 参数说明及获取方式 参数说明 ak：账号 Access Key sk：账号 Secret Access Key 认证用的ak和sk硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放, 使用时解密, 确保安全。 本示例以ak和sk保存在环境变量中为例，运行本示例前请先在本地环境中设置环境变量xxx_SDK_AK和xxx_SDK_SK。 regionName ：所属区域名称 projectId ：项目ID 通过以下方式可获取AK/SK，项目ID及对应的region信息。 登录管理控制台。 鼠标指向界面右上角的登录用户名，在下拉列表中单击“我的凭证”。 在左侧导航栏中选择“访问密钥”，单击“新增访问密钥”。根据提示输入对应信息，单击“确定”。 在弹出的提示页面单击“立即下载”。下载成功后，打开凭证文件，获取AK/SK信息。 左侧导航栏单击“API凭证”，在“项目列表”中获取“项目ID”即为project_id值，对应的“项目”即为region的值。

Token认证方式样例代码 代码样例 1 2 3 4 5 6 7 String domainName = "domainname"; String userName = "username"; String password = "password"; String regionName = "regionname"; String projectId = "project_id"; DLIInfo dliInfo = new DLIInfo(regionName, domainName, userName, password, projectId); DLIClient client = new DLIClient(AuthenticationMode.TOKEN, dliInfo); 参数说明 参数获取方式请参考获取账号、 IAM 用户、项目、用户组、区域、委托的名称和ID。 domainname：账号名。 username：用户名 password：用户名密码 regionname：所属区域名称 project_id：项目ID 认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。 可以通过set方式修改endpoint，即dliInfo.setServerEndpoint(endpoint)。

提交作业 DLI提供提交作业和查询作业的接口。您可以通过提交接口提交作业，如果需要查询结果可以调用查询接口查询该作业的结果。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 //实例化SQLJob对象，传入执行SQL所需的queue,数据库名，SQL语句 private static void runSqlJob(Queue queue, Table obsTable) throws DLIException { String sql = "select * from " + obsTable.getTableName(); String queryResultPath = "OBS Path"; SQLJob sqlJob = new SQLJob(queue, obsTable.getDb().getDatabaseName(), sql); System.out.println("start submit SQL job..."); //调用SQLJob对象的submit接口提交查询作业 sqlJob.submit(); //调用SQLJob对象的getStatus接口查询作业状态 JobStatus status = sqlJob.getStatus(); System.out.println(status); System.out.println("start export Result..."); //调用SQLJob对象的exportResult接口导出查询结果，其中queryResultPath为导出数据的路径 sqlJob.exportResult(queryResultPath, StorageType. CS V, CompressType.GZIP, ExportMode.ERRORIFEXISTS, null); System.out.println("Job id: " + sqlJob.getJobId() + ", Status : " + status.getName()); }

查询作业结果 DLI提供查询作业结果的接口。您可以使用该接口通过JobId查询该作业信息。示例代码如下： 1 2 3 4 5 6 7 8 9 private static void getJobResultInfo(DLIClient client) throws DLIException { String jobId = "4c4f7168-5bc4-45bd-8c8a-43dfc85055d0"; JobResultInfo jobResultInfo = client.queryJobResultInfo(jobId); //查询job信息 System.out.println(jobResultInfo.getJobId()); System.out.println(jobResultInfo.getDetail()); System.out.println(jobResultInfo.getJobStatus()); System.out.println(jobResultInfo.getJobType()); }

导出查询结果 DLI提供导出查询结果的接口。您可以使用该接口导出当前工程下，在编辑框中提交的查询作业的结果。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 //实例化SQLJob对象，传入执行SQL所需的queue,数据库名，SQL语句 private static void exportSqlResult(Queue queue, Table obsTable) throws DLIException { String sql = "select * from " + obsTable.getTableName(); String queryResultPath = "OBS Path"; SQLJob sqlJob = new SQLJob(queue, obsTable.getDb().getDatabaseName(), sql); System.out.println("start submit SQL job..."); //调用SQLJob对象的submit接口提交查询作业 sqlJob.submit(); //调用SQLJob对象的getStatus接口查询作业状态 JobStatus status = sqlJob.getStatus(); System.out.println(status); System.out.println("start export Result..."); //调用SQLJob对象的exportResult接口导出查询结果，其中exportPath为导出数据的路径,JSON为导出格式，queueName为执行导出作业的队列，limitNum为导出作业结果条数，0表示全部导出 sqlJob.exportResult(exportPath + "result", StorageType.JSON, CompressType.NONE, ExportMode.ERRORIFEXISTS, queueName, true, 5); }

导入数据 DLI提供导入数据的接口。您可以使用该接口将存储在OBS中的数据导入到已创建的DLI表或者OBS表中。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 //实例化importJob对象，构造函数的入参包括队列、数据库名、表名（通过实例化Table对象获取）和数据路径 private static void importData(Queue queue, Table DLITable) throws DLIException { String dataPath = "OBS Path"; queue = client.getQueue("queueName"); CsvFormatInfo formatInfo = new CsvFormatInfo(); formatInfo.setWithColumnHeader(true); formatInfo.setDelimiter(","); formatInfo.setQuoteChar("\""); formatInfo.setEscapeChar("\\"); formatInfo.setDateFormat("yyyy/MM/dd"); formatInfo.setTimestampFormat("yyyy-MM-dd HH:mm:ss"); String dbName = DLITable.getDb().getDatabaseName(); String tableName = DLITable.getTableName(); ImportJob importJob = new ImportJob(queue, dbName, tableName, dataPath); importJob.setStorageType(StorageType.CSV); importJob.setCsvFormatInfo(formatInfo); System.out.println("start submit import table: " + DLITable.getTableName()); //调用ImportJob对象的submit接口提交导入作业 importJob.submit(); //调用ImportJob对象的getStatus接口查询导入作业状态 JobStatus status = importJob.getStatus(); System.out.println("Job id: " + importJob.getJobId() + ", Status : " + status.getName()); } 在提交导入作业前，可选择设置导入数据的格式，如样例所示，调用ImportJob对象的setStorageType接口设置数据存储类型为csv，数据的具体格式通过调用ImportJob对象的setCsvFormatInfo接口进行设置。 在提交导入作业前，可选择设置导入数据的分区并配置是否是overwrite写入，分区信息可以调用ImportJob对象的setPartitionSpec接口设置，如：importJob.setPartitionSpec(new PartitionSpec("part1=value1,part2=value2"))，也可以在创建ImportJob对象的时候直接通过参数的形式创建 。导入作业默认是追加写，如果需要覆盖写，则可以调用ImportJob对象的setOverWrite接口设置，如：importJob.setOverWrite(Boolean.TRUE)。 当OBS桶目录下有文件夹和文件同名时，加载数据会优先指向该路径下的文件而非文件夹。建议创建OBS对象时，在同一级中不要出现同名的文件和文件夹。

导出数据 DLI提供导出数据的接口。您可以使用该接口将DLI表中的数据导出到OBS中。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 //实例化ExportJob对象，传入导出数据所需的队列、数据库名、表名（通过实例化Table对象获取）和导出数据的存储路径，仅支持Table类型为MANAGED private static void exportData(Queue queue, Table DLITable) throws DLIException { String dataPath = "OBS Path"; queue = client.getQueue("queueName"); String dbName = DLITable.getDb().getDatabaseName(); String tableName = DLITable.getTableName(); ExportJob exportJob = new ExportJob(queue, dbName, tableName, dataPath); exportJob.setStorageType(StorageType.CSV); exportJob.setCompressType(CompressType.GZIP); exportJob.setExportMode(ExportMode.ERRORIFEXISTS); System.out.println("start export DLI Table data..."); //调用ExportJob对象的submit接口提交导出作业 exportJob.submit(); //调用ExportJob对象的getStatus接口查询导出作业状态 JobStatus status = exportJob.getStatus(); System.out.println("Job id: " + exportJob.getJobId() + ", Status : " + status.getName()); } 在提交导出作业前，可选设置数据格式，压缩类型，导出模式等，如样例所示，分别调用ExportJob对象的setStorageType、setCompressType、setExportMode接口设置，其中setStorageType仅支持csv格式。 当OBS桶目录下有文件夹和文件同名时，加载数据会优先指向该路径下的文件而非文件夹。建议创建OBS对象时，在同一级中不要出现同名的文件和文件夹。

导入分区数据 DLI提供导入数据的接口。您可以使用该接口将存储在OBS中的数据导入到已创建的DLI表或者OBS表指定分区中。示例代码如下： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 //实例化importJob对象，构造函数的入参包括队列、数据库名、表名（通过实例化Table对象获取）和数据路径 private static void importData(Queue queue, Table DLITable) throws DLIException { String dataPath = "OBS Path"; queue = client.getQueue("queueName"); CsvFormatInfo formatInfo = new CsvFormatInfo(); formatInfo.setWithColumnHeader(true); formatInfo.setDelimiter(","); formatInfo.setQuoteChar("\""); formatInfo.setEscapeChar("\\"); formatInfo.setDateFormat("yyyy/MM/dd"); formatInfo.setTimestampFormat("yyyy-MM-dd HH:mm:ss"); String dbName = DLITable.getDb().getDatabaseName(); String tableName = DLITable.getTableName(); PartitionSpec partitionSpec = new PartitionSpec("part1=value1,part2=value2"); Boolean isOverWrite = true; ImportJob importJob = new ImportJob(queue, dbName, tableName, dataPath, partitionSpec, isOverWrite); importJob.setStorageType(StorageType.CSV); importJob.setCsvFormatInfo(formatInfo); System.out.println("start submit import table: " + DLITable.getTableName()); //调用ImportJob对象的submit接口提交导入作业 importJob.submit(); //调用ImportJob对象的getStatus接口查询导入作业状态 JobStatus status = importJob.getStatus(); System.out.println("Job id: " + importJob.getJobId() + ", Status : " + status.getName()); } 在创建ImportJob对象的时候分区信息PartitionSpec也可以直接传入分区字符串。 partitionSpec如果导入时指定部分列为分区列，而导入的数据只包含了指定的分区信息，则数据导入后的未指定的分区列字段会存在null值等异常值。 示例中isOverWrite表示是否是覆盖写，为true表示覆盖写，为false表示追加写。目前不支持overwrite覆盖写整表，只支持overwrite写指定分区。如果需要追加写指定分区，则在创建ImportJob的时候指定isOverWrite为false。

DLI SDK简介 数据湖探索 服务软件开发工具包（DLI SDK，Data Lake Insight Software Development Kit）是对DLI服务提供的REST API进行的作业提交的封装，以简化用户的开发工作。用户直接调用DLI SDK提供的接口函数即可实现使用提交DLI SQL和DLI Spark作业。 DLI支持的SDK分为SDK V3和DLI服务自行开发的SDK。 （推荐）DLI SDK V3：是根据定义API的YAML文件统一自动生成，其接口参数与服务的API一致。 具体操作请参考SDK V3版本开发指南。 DLI SDK（服务自研）：是DLI服务自行开发的SDK，本手册介绍DLI 自研SDK的使用方法。相关开发包请从华为云DLI 开发工具包（SDK）获取。 DLI SDK V2版本操作指导： Java SDK操作指导请参考Java SDK（DLI SDK V2）。 Python SDK操作指导请参考Python SDK（DLI SDK V2） DLI SDK V1版本操作指导（不推荐使用）： Java SDK操作指导请参考Java SDK（DLI SDK V1） Python SDK操作指导请参考Python SDK（DLI SDK V1）

Token认证方式样例代码 代码样例 1 2 3 4 5 6 7 8 9 def init_token_dli_client(): auth_mode = 'token' region = 'xxx' project_id = 'xxxx' account = 'xxx account' user = 'xxxx' password = 'xxxx' dli_client = DliClient(auth_mode=auth_mode, region=region, project_id=project_id,account=account, user=user, password=password) return dli_client 参数说明 参数获取方式请参考获取账号、IAM用户、项目、用户组、区域、委托的名称和ID。 domainname：账号名。 username：用户名 password：用户名密码 regionname：所属区域名称 project_id：项目ID 认证用的password硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。 可以通过set方式修改endpoint，即dliInfo.setServerEndpoint(endpoint)。

AK/SK认证方式样例代码 代码样例 1 2 3 4 5 6 7 8 def init_aksk_dli_client(): auth_mode = 'aksk' region = 'xxx' project_id = 'xxxx' ak = System.getenv("xxx_SDK_AK")//访问密钥ID。 sk = System.getenv("xxx_SDK_SK")//与访问密钥ID结合使用的密钥。 dli_client = DliClient(auth_mode=auth_mode, region=region, project_id=project_id,ak=ak, sk=sk) return dli_client 参数说明与获取方式 参数说明 ak：账号 Access Key sk：账号 Secret Access Key 认证用的ak和sk硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放, 使用时解密, 确保安全。 本示例以ak和sk保存在环境变量中为例，运行本示例前请先在本地环境中设置环境变量xxx_SDK_AK和xxx_SDK_SK。 regionName ：所属区域名称 projectId ：项目ID 通过以下方式可获取AK/SK，项目ID及对应的region信息。 登录管理控制台。 鼠标指向界面右上角的登录用户名，在下拉列表中单击“我的凭证”。 在左侧导航栏中选择“访问密钥”，单击“新增访问密钥”。根据提示输入对应信息，单击“确定”。 在弹出的提示页面单击“立即下载”。下载成功后，打开凭证文件，获取AK/SK信息。 左侧导航栏单击“API凭证”，在“项目列表”中获取“项目ID”即为project_id值，对应的“项目”即为region的值。

使用须知 要使用DLI Java SDK 访问指定服务的 API ，您需要确认已在DLI控制台开通当前服务并完成服务授权。 Java SDK 支持 Java JDK 1.8 及其以上版本。关于Java开发环境的配置请参考Java SDK环境配置。 关于Java SDK的获取与安装请参考Java SDK的获取与安装。 使用SDK工具访问DLI，需要用户初始化DLI客户端。用户可以使用AK/SK(Access Key ID/Secret Access Key)或Token两种认证方式初始化客户端，具体操作请参考初始化DLI客户端