当前位置:

首页 > 编程开发 > SpringBoot怎么整合dataworks

SpringBoot怎么整合dataworks

注意事项这里测试主要是调用拉取dataworks上拉取的脚本,并存储到本地。脚本包含两部分1、开发的odps脚本(通过OpenApi获取)2、建表语句脚本(通过dataworks信息去连接maxCompute获取建立语句)阿里云Dataworks的openApi分页查询限制,一次最多查询100条。我们拉取脚本需要分多页查询该项目使用到了MaxCompute的SDK/JDBC方式连接,SpringBoot操作MaxComputeSDK/JDBC连接整合实现实现主要是编写工具类,如果需要则可以配置成Sprin

    注意事项

    这里测试主要是调用拉取dataworks上拉取的脚本,并存储到本地。
    脚本包含两部分

    1、开发的odps脚本(通过OpenApi获取)2、建表语句脚本(通过dataworks信息去连接maxCompute获取建立语句)

    阿里云Dataworks的openApi分页查询限制,一次最多查询100条。我们拉取脚本需要分多页查询

    该项目使用到了MaxCompute的SDK/JDBC方式连接,SpringBoot操作MaxCompute SDK/JDBC连接

    整合实现

    实现主要是编写工具类,如果需要则可以配置成SpringBean,注入容器即可使用

    依赖引入

    
        1.8
        
        0.40.8-public
        
        3.0.1
        
        3.4.2
        4.5.20
    
    
    
        org.springframework.boot
        spring-boot-starter-web
    
    
        org.springframework.boot
        spring-boot-configuration-processor
        true
    
    
        org.projectlombok
        lombok
        true
    
    
    
        com.aliyun.odps
        odps-sdk-core
        ${max-compute-sdk.version}
    
    
    
        com.aliyun.odps
        odps-jdbc
        ${max-compute-jdbc.version}
        jar-with-dependencies
    
    
    
        com.aliyun
        aliyun-java-sdk-core
        ${aliyun-java-sdk.version}
    
    
        com.aliyun
        aliyun-java-sdk-dataworks-public
        ${dataworks-sdk.version}
    
    

    请求参数类编写

    /**
     * @Description
     * @Author itdl
     * @Date 2022/08/09 15:12
     */
    @Data
    public class DataWorksOpenApiConnParam {
        /**
         * 区域 eg. cn-shanghai
         */
        private String region;
    
        /**
         * 访问keyId
         */
        private String aliyunAccessId;
        /**
         * 密钥
         */
        private String aliyunAccessKey;
    
        /**
         * 访问端点  就是API的URL前缀
         */
        private String endPoint;
    
        /**
         * 数据库类型 如odps
         */
        private String datasourceType;
    
        /**
         * 所属项目
         */
        private String project;
    
        /**
         * 项目环境 dev  prod
         */
        private String projectEnv;
    }

    工具类编写

    基础类准备,拉取脚本之后的回调函数

    为什么需要回调函数,因为拉取的是所有脚本,如果合并每次分页结果的话,会导致内存溢出,而使用回调函数只是每次循环增加处理函数

    /**
     * @Description
     * @Author itdl
     * @Date 2022/08/09 15:12
     */
    @Data
    public class DataWorksOpenApiConnParam {
        /**
         * 区域 eg. cn-shanghai
         */
        private String region;
    
        /**
         * 访问keyId
         */
        private String aliyunAccessId;
        /**
         * 密钥
         */
        private String aliyunAccessKey;
    
        /**
         * 访问端点  就是API的URL前缀
         */
        private String endPoint;
    
        /**
         * 数据库类型 如odps
         */
        private String datasourceType;
    
        /**
         * 所属项目
         */
        private String project;
    
        /**
         * 项目环境 dev  prod
         */
        private String projectEnv;
    }

    初始化操作

    主要是实例化dataworks openApi接口的客户端信息,maxCompute连接的工具类初始化(包括JDBC,SDK方式)

    private static final String MAX_COMPUTE_JDBC_URL_FORMAT = "http://service.%s.maxcompute.aliyun.com/api";
    /**默认的odps接口地址 在Odps中也可以看到该变量*/
    private static final String defaultEndpoint = "http://service.odps.aliyun.com/api";
    /**
     * dataworks连接参数
     *
     */
    private final DataWorksOpenApiConnParam connParam;
    
    /**
     * 可以使用dataworks去连接maxCompute 如果连接的引擎是maxCompute的话
     */
    private final MaxComputeJdbcUtil maxComputeJdbcUtil;
    
    private final MaxComputeSdkUtil maxComputeSdkUtil;
    
    private final boolean odpsSdk;
    
    
    /**
     * 客户端
     */
    private final IAcsClient client;
    
    public DataWorksOpenApiUtil(DataWorksOpenApiConnParam connParam, boolean odpsSdk) {
        this.connParam = connParam;
        this.client = buildClient();
        this.odpsSdk = odpsSdk;
        if (odpsSdk){
            this.maxComputeJdbcUtil = null;
            this.maxComputeSdkUtil = buildMaxComputeSdkUtil();
        }else {
            this.maxComputeJdbcUtil = buildMaxComputeJdbcUtil();
            this.maxComputeSdkUtil = null;
        }
    }
    
    private MaxComputeSdkUtil buildMaxComputeSdkUtil() {
        final MaxComputeSdkConnParam param = new MaxComputeSdkConnParam();
    
        // 设置账号密码
        param.setAliyunAccessId(connParam.getAliyunAccessId());
        param.setAliyunAccessKey(connParam.getAliyunAccessKey());
    
        // 设置endpoint
        param.setMaxComputeEndpoint(defaultEndpoint);
    
        // 目前只处理odps的引擎
        final String datasourceType = connParam.getDatasourceType();
        if (!"odps".equals(datasourceType)){
            throw new BizException(ResultCode.DATA_WORKS_ENGINE_SUPPORT_ERR);
        }
    
        // 获取项目环境,根据项目环境连接不同的maxCompute
        final String projectEnv = connParam.getProjectEnv();
    
        if ("dev".equals(projectEnv)){
            // 开发环境dataworks + _dev就是maxCompute的项目名
            param.setProjectName(String.join("_", connParam.getProject(), projectEnv));
        }else {
            // 生产环境dataworks的项目名和maxCompute一致
            param.setProjectName(connParam.getProject());
        }
    
        return new MaxComputeSdkUtil(param);
    }
    
    private MaxComputeJdbcUtil buildMaxComputeJdbcUtil() {
        final MaxComputeJdbcConnParam param = new MaxComputeJdbcConnParam();
    
        // 设置账号密码
        param.setAliyunAccessId(connParam.getAliyunAccessId());
        param.setAliyunAccessKey(connParam.getAliyunAccessKey());
    
        // 设置endpoint
        param.setEndpoint(String.format(MAX_COMPUTE_JDBC_URL_FORMAT, connParam.getRegion()));
    
        // 目前只处理odps的引擎
        final String datasourceType = connParam.getDatasourceType();
        if (!"odps".equals(datasourceType)){
            throw new BizException(ResultCode.DATA_WORKS_ENGINE_SUPPORT_ERR);
        }
    
        // 获取项目环境,根据项目环境连接不同的maxCompute
        final String projectEnv = connParam.getProjectEnv();
    
        if ("dev".equals(projectEnv)){
            // 开发环境dataworks + _dev就是maxCompute的项目名
            param.setProjectName(String.join("_", connParam.getProject(), projectEnv));
        }else {
            // 生产环境dataworks的项目名和maxCompute一致
            param.setProjectName(connParam.getProject());
        }
    
        return new MaxComputeJdbcUtil(param);
    }

    调用OpenApi拉取所有脚本

    /**
     * 根据文件夹路径分页查询该路径下的文件(脚本)
     * @param pageSize 每页查询多少数据
     * @param folderPath 文件所在目录
     * @param userType 文件所属功能模块 可不传
     * @param fileTypes 设置文件代码类型 逗号分割 可不传
     */
    public void listAllFiles(Integer pageSize, String folderPath, String userType, String fileTypes, CallBack.FileCallBack callBack) throws ClientException {
        pageSize = setPageSize(pageSize);
        // 创建请求
        final ListFilesRequest request = new ListFilesRequest();
    
        // 设置分页参数
        request.setPageNumber(1);
        request.setPageSize(pageSize);
    
        // 设置上级文件夹
        request.setFileFolderPath(folderPath);
    
        // 设置区域和项目名称
        request.setSysRegionId(connParam.getRegion());
        request.setProjectIdentifier(connParam.getProject());
    
        // 设置文件所属功能模块
        if (!ObjectUtils.isEmpty(userType)){
            request.setUseType(userType);
        }
        // 设置文件代码类型
        if (!ObjectUtils.isEmpty(fileTypes)){
            request.setFileTypes(fileTypes);
        }
    
        // 发起请求
        ListFilesResponse res = client.getAcsResponse(request);
    
        // 获取分页总数
        final Integer totalCount = res.getData().getTotalCount();
        // 返回结果
        final List resultList = res.getData().getFiles();
        // 计算能分几页
        long pages = totalCount % pageSize == 0 ? (totalCount / pageSize) : (totalCount / pageSize) + 1;
        // 只有1页 直接返回
        if (pages <= 1){
            callBack.handle(resultList);
            return;
        }
    
        // 第一页执行回调
        callBack.handle(resultList);
    
        // 分页数据 从第二页开始查询 同步拉取,可以优化为多线程拉取
        for (int i = 2; i <= pages; i++) {
            //第1页
            request.setPageNumber(i);
            //每页大小
            request.setPageSize(pageSize);
            // 发起请求
            res = client.getAcsResponse(request);
            final List tableEntityList = res.getData().getFiles();
            if (!ObjectUtils.isEmpty(tableEntityList)){
                // 执行回调函数
                callBack.handle(tableEntityList);
            }
        }
    }

    内部连接MaxCompute拉取所有DDL脚本内容

    DataWorks工具类代码,通过回调函数处理

        /**
         * 获取所有的DDL脚本
         * @param callBack 回调处理函数
         */
        public void listAllDdl(CallBack.DdlCallBack callBack){
            if (odpsSdk){
                final List tableInfos = maxComputeSdkUtil.getTableInfos();
                for (TableMetaInfo tableInfo : tableInfos) {
                    final String tableName = tableInfo.getTableName();
                    final String sqlCreateDesc = maxComputeSdkUtil.getSqlCreateDesc(tableName);
                    callBack.handle(tableName, sqlCreateDesc);
                }
            }
        }

    MaxCompute工具类代码,根据表名获取建表语句, 以SDK为例, JDBC直接执行show create table即可拿到建表语句

    /**
     * 根据表名获取建表语句
     * @param tableName 表名
     * @return
     */
    public String getSqlCreateDesc(String tableName) {
        final Table table = odps.tables().get(tableName);
        // 建表语句
        StringBuilder mssqlDDL = new StringBuilder();
    
        // 获取表结构
        TableSchema tableSchema = table.getSchema();
        // 获取表名表注释
        String tableComment = table.getComment();
    
        //获取列名列注释
        List columns = tableSchema.getColumns();
        /*组装成mssql的DDL*/
        // 表名
        mssqlDDL.append("CREATE TABLE IF NOT EXISTS ");
        mssqlDDL.append(tableName).append("\n");
        mssqlDDL.append(" (\n");
        //列字段
        int index = 1;
        for (Column column : columns) {
            mssqlDDL.append("  ").append(column.getName()).append("\t\t").append(column.getTypeInfo().getTypeName());
            if (!ObjectUtils.isEmpty(column.getComment())) {
                mssqlDDL.append(" COMMENT '").append(column.getComment()).append("'");
            }
            if (index == columns.size()) {
                mssqlDDL.append("\n");
            } else {
                mssqlDDL.append(",\n");
            }
            index++;
        }
        mssqlDDL.append(" )\n");
        //获取分区
        List partitionColumns = tableSchema.getPartitionColumns();
        int partitionIndex = 1;
        if (!ObjectUtils.isEmpty(partitionColumns)) {
            mssqlDDL.append("PARTITIONED BY (");
        }
        for (Column partitionColumn : partitionColumns) {
            final String format = String.format("%s %s COMMENT '%s'", partitionColumn.getName(), partitionColumn.getTypeInfo().getTypeName(), partitionColumn.getComment());
            mssqlDDL.append(format);
            if (partitionIndex == partitionColumns.size()) {
                mssqlDDL.append("\n");
            } else {
                mssqlDDL.append(",\n");
            }
            partitionIndex++;
        }
    
        if (!ObjectUtils.isEmpty(partitionColumns)) {
            mssqlDDL.append(")\n");
        }
    //        mssqlDDL.append("STORED AS ALIORC  \n");
    //        mssqlDDL.append("TBLPROPERTIES ('comment'='").append(tableComment).append("');");
        mssqlDDL.append(";");
        return mssqlDDL.toString();
    }

    测试代码

    public static void main(String[] args) throws ClientException {
        final DataWorksOpenApiConnParam connParam = new DataWorksOpenApiConnParam();
        connParam.setAliyunAccessId("您的阿里云账号accessId");
        connParam.setAliyunAccessKey("您的阿里云账号accessKey");
        // dataworks所在区域
        connParam.setRegion("cn-chengdu");
        // dataworks所属项目
        connParam.setProject("dataworks所属项目");
        // dataworks所属项目环境 如果不分环境的话设置为生产即可
        connParam.setProjectEnv("dev");
        // 数据引擎类型 odps
        connParam.setDatasourceType("odps");
        // ddataworks接口地址
        connParam.setEndPoint("dataworks.cn-chengdu.aliyuncs.com");
        final DataWorksOpenApiUtil dataWorksOpenApiUtil = new DataWorksOpenApiUtil(connParam, true);
    
        // 拉取所有ODPS脚本
        dataWorksOpenApiUtil.listAllFiles(100, "", "", "10", files -> {
            // 处理文件
            for (ListFilesResponse.Data.File file : files) {
                final String fileName = file.getFileName();
                System.out.println(fileName);
            }
        });
    
        // 拉取所有表的建表语句
        dataWorksOpenApiUtil.listAllDdl((tableName, tableDdlContent) -> {
            System.out.println("=======================================");
            System.out.println("表名:" + tableName + "内容如下:\n");
            System.out.println(tableDdlContent);
            System.out.println("=======================================");
        });
    }

    测试结果

    test_001脚本
    test_002脚本
    test_003脚本
    test_004脚本
    test_005脚本
    =======================================
    表名:test_abc_info内容如下:

    CREATE TABLE IF NOT EXISTS test_abc_info
     (
        test_abc1        STRING COMMENT '字段1',
        test_abc2        STRING COMMENT '字段2',
        test_abc3        STRING COMMENT '字段3',
        test_abc4        STRING COMMENT '字段4',
        test_abc5        STRING COMMENT '字段5',
        test_abc6        STRING COMMENT '字段6',
        test_abc7        STRING COMMENT '字段7',
        test_abc8        STRING COMMENT '字段8'
     )
    PARTITIONED BY (p_date STRING COMMENT '数据日期'
    )
    ;
    =======================================
    Disconnected from the target VM, address: '127.0.0.1:59509', transport: 'socket'

    本文内容来源于互联网,如有侵权请联系删除。
    作者最新文章
    编程开发
    相关文章 更多
    using namespace 使用中遇到的问题怎么解决
    using namespace 使用中遇到的问题怎么解决

    命名空间的基本概念与常见引入问题在C++等编程语言中,命名空间(namespace)是一种将代码标识符(如变量、函数、类名)封装在特定名称下的机制,其主要目的是避免命名冲突,尤其是在大型项目或使用多个第三方库时。使用“using namespace”指令可以将指定命名空间中的所有名称引入当前作用域,

    c语言函数递归 实操经验总结:这些技巧很实用
    c语言函数递归 实操经验总结:这些技巧很实用

    理解递归的基本原理在C语言中,递归是一种函数调用自身的编程技术。要掌握它,首先需要理解其核心思想:将一个复杂的大问题,分解为一个或几个与原问题相似但规模更小的子问题,直到子问题足够简单,可以直接求解。这个过程通常包含两个关键部分:递归出口和递归体。递归出口定义了问题何时不再继续分解,即最简单、可直接

    c语言函数递归 怎么选?常见方案对比分析
    c语言函数递归 怎么选?常见方案对比分析

    递归函数的基本概念与适用场景在C语言编程中,递归是一种函数调用自身的编程技巧。它并非适用于所有问题,但在处理某些具有自相似结构的问题时,能提供极其清晰和优雅的解决方案。递归的核心思想是将一个大规模问题分解为一个或多个同类型但规模更小的子问题,直到子问题简单到可以直接求解。典型的适用场景包括树形结构的

    Objective-C 内存管理入门:从 alloc 到 dealloc 的生命周期详解
    Objective-C 内存管理入门:从 alloc 到 dealloc 的生命周期详解

    理解内存管理的基石在Objective-C的编程世界中,内存管理是开发者必须掌握的核心技能之一。它直接关系到应用的性能、稳定性与资源利用效率。与一些采用自动垃圾回收机制的语言不同,Objective-C在很长一段时间里,依赖一套基于引用计数的、需要开发者部分介入的管理规则。这套规则的核心思想是明确的

    如何正确使用 dealloc 以避免 iOS 应用中的内存泄漏
    如何正确使用 dealloc 以避免 iOS 应用中的内存泄漏

    理解 dealloc 的角色与时机在 iOS 应用开发中,内存管理是保障应用性能与稳定性的基石。dealloc 方法是 Objective-C 中对象生命周期结束时的关键回调,它标志着对象即将被系统回收内存。正确理解其触发时机至关重要:当一个对象的引用计数降为零时,运行时系统会自动调用该对象的 de

    深入理解 Objective-C 中的 dealloc 方法:内存管理核心机制
    深入理解 Objective-C 中的 dealloc 方法:内存管理核心机制

    内存管理的基石在Objective-C的世界里,内存管理是开发者必须掌握的核心技能之一。作为一门在手动引用计数(MRC)时代诞生的语言,Objective-C要求程序员对对象的生命周期有清晰的认识。dealloc方法正是这一生命周期中至关重要的终点站。它是一个实例方法,当对象的引用计数降为零时,系统

    理解 native2ascii:Java 国际化开发中的字符编码工具
    理解 native2ascii:Java 国际化开发中的字符编码工具

    native2ascii 工具的基本定位在Ja va应用程序的国际化与本地化开发过程中,处理非拉丁字符集是一个常见且关键的环节。Ja va内部使用Unicode字符集来统一表示全球各种语言的文字,但其属性文件(.properties)在历史上要求使用ASCII编码,或者更准确地说,要求非ASCII字

    如何使用 native2ascii 转换中文字符为 Unicode 转义序列
    如何使用 native2ascii 转换中文字符为 Unicode 转义序列

    理解 native2ascii 工具的基本用途在软件开发,特别是涉及国际化处理的场景中,开发者常常需要处理不同编码的文本资源。native2ascii 是 Ja va 开发工具包(JDK)中提供的一个命令行实用程序,其主要功能是将包含本地字符编码(非ASCII字符)的文件,转换为包含 Unicode

    Java native2ascii 命令详解:解决属性文件乱码问题
    Java native2ascii 命令详解:解决属性文件乱码问题

    native2ascii 命令的由来与作用在Ja va开发中,处理国际化资源文件是一个常见需求。资源文件通常以.properties格式存储,用于支持多语言界面。然而,Ja va属性文件默认采用ISO-8859-1字符集编码,这导致了一个直接的问题:当文件中包含非拉丁字符(如中文、日文、韩文等)时,

    一个 memwatch 实战案例:定位野指针问题
    一个 memwatch 实战案例:定位野指针问题

    内存监控工具的价值与挑战在软件开发,尤其是使用C/C++这类手动管理内存的语言时,内存错误是程序员最常遭遇的难题之一。其中,野指针问题因其隐蔽性和破坏性,往往成为最难定位的“幽灵”缺陷。它可能潜伏在代码中,在特定条件下才被触发,导致程序崩溃、数据损坏或难以预测的行为。传统的调试手段,如打印日志或使用

    查看更多
    精品专题 更多
    装机必备
    装机必备

    正软商城装机必备专区,精选办公、浏览器、安全防护、影音播放、压缩解压、设计创作和系统工具等电脑常用正版软件,帮助用户快速完成新电脑软件配置。

    Windows
    Windows

    正软商城Windows软件专区,汇集适用于Windows电脑的办公、设计、安全防护、影音播放、开发工具和系统优化软件,提供软件介绍、系统要求、正版授权及购买下载服务。

    macOS软件
    macOS软件

    正软商城macOS软件专区,精选适用于Mac电脑的办公、设计、影音、效率、开发和系统工具,提供软件功能介绍、macOS兼容版本、正版授权及购买下载服务。

    Mac软件 更多
    灵活计算器
    灵活计算器
    macOS/iOS/Android

    灵活计算器是一款笔记式算数应用,支持实时计算、动态关联和云端同步功能。记录、整理和输出之间的过渡会更自然,适合长期写作、做笔记或持续沉淀个人内容。

    赤友清理大师
    赤友清理大师
    macOS

    赤友清理大师是一款为 Mac 设计的智能清理优化工具,可精准扫描垃圾、大文件、重复文件等,释放磁盘空间。做扫描整理、文字提取和表格转换时,它能把识别后的处理步骤接得更顺,资料录入这类场景会省下不少时间。

    极度公式
    极度公式
    Windows/macOS/Linux

    极度公式是一款跨平台专业LaTeX公式识别编辑软件,支持OCR公式识别和多平台编辑。和使用说明,避免使用,享受完整功能与稳定支持。做扫描整理、文字提取和表格转换时,它能把识别后的处理步骤接得更顺,资料录入这类场景会省下不少时间。

    WINDOWS 更多
    Windows 10
    Windows 10
    Windows

    Windows 10 是一款微软推出的经典操作系统,拥有硬件兼容性与多任务处理能力。它更偏向把系统状态查看和常用调节动作放在一起,适合需要持续观察和微调设备状态的场景。

    极度公式
    极度公式
    Windows/macOS/Linux

    极度公式是一款跨平台专业LaTeX公式识别编辑软件,支持OCR公式识别和多平台编辑。和使用说明,避免使用,享受完整功能与稳定支持。做扫描整理、文字提取和表格转换时,它能把识别后的处理步骤接得更顺,资料录入这类场景会省下不少时间。

    密码键盘
    密码键盘
    Windows/macOS/iOS/Android

    密码键盘是一款兼具安全性与便捷性的高效密码管理器。日常使用里的持续防护和信息管理会更突出,适合把安全控制放进长期使用流程中的场景。