SpringBoot整合dataworks

注意事項

阿裡雲的dataworks提供了OpenApi, 需要是企業版或旗艦版才能夠調用，也就是付費項目。

這裡測試主要是調用拉取dataworks上拉取的腳本，并存儲到本地。腳本包含兩部分

1、開發的odps腳本(通過OpenApi擷取)
2、建表語句腳本(通過dataworks資訊去連接配接maxCompute擷取建立語句)

阿裡雲Dataworks的openApi分頁查詢限制，一次最多查詢100條。我們拉取腳本需要分多頁查詢

該項目使用到了MaxCompute的SDK/JDBC方式連接配接，SpringBoot操作MaxCompute SDK/JDBC連接配接

整合實作

實作主要是編寫工具類，如果需要則可以配置成SpringBean，注入容器即可使用

依賴引入

<properties>
    <java.version>1.8</java.version>
    <!--maxCompute-sdk-版本号-->
    <max-compute-sdk.version>0.40.8-public</max-compute-sdk.version>
    <!--maxCompute-jdbc-版本号-->
    <max-compute-jdbc.version>3.0.1</max-compute-jdbc.version>
    <!--dataworks版本号-->
    <dataworks-sdk.version>3.4.2</dataworks-sdk.version>
    <aliyun-java-sdk.version>4.5.20</aliyun-java-sdk.version>
</properties>
<dependencies>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-configuration-processor</artifactId>
    <optional>true</optional>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>
<!--max compute sdk-->
<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-sdk-core</artifactId>
    <version>${max-compute-sdk.version}</version>
</dependency>
<!--max compute jdbc-->
<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-jdbc</artifactId>
    <version>${max-compute-jdbc.version}</version>
    <classifier>jar-with-dependencies</classifier>
</dependency>
<!--dataworks需要引入aliyun-sdk和dataworks本身-->
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-core</artifactId>
    <version>${aliyun-java-sdk.version}</version>
</dependency>
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-dataworks-public</artifactId>
    <version>${dataworks-sdk.version}</version>
</dependency>
</dependencies>

請求參數類編寫

/**
 * @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;
}

工具類編寫

基礎類準備，拉取腳本之後的回調函數

為什麼需要回調函數，因為拉取的是所有腳本，如果合并每次分頁結果的話，會導緻記憶體溢出，而使用回調函數隻是每次循環增加處理函數

public static class CallBack {
        public interface FileCallBack {
            /**
             * 每次分頁查詢檔案夾回調
             */
            void handle(List<ListFilesResponse.Data.File> files);
        }


        public interface DdlCallBack {
            /**
             * 擷取建表語句回調結果
             */
            void handle(String tableName, String tableDdlContent);
        }
    }

初始化操作

主要是執行個體化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<ListFilesResponse.Data.File> 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<ListFilesResponse.Data.File> 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<TableMetaInfo> 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<Column> 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<Column> 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
 (
       STRING COMMENT '字段1',
       STRING COMMENT '字段2',
       STRING COMMENT '字段3',
       STRING COMMENT '字段4',
       STRING COMMENT '字段5',
       STRING COMMENT '字段6',
       STRING COMMENT '字段7',
       STRING COMMENT '字段8'
 )
PARTITIONED BY (p_date STRING COMMENT '資料日期'
)
;
=======================================
Disconnected from the target VM, address: '127.0.0.1:59509', transport: 'socket'

SpringBoot整合dataworks

注意事項

整合實作

依賴引入

請求參數類編寫

工具類編寫

基礎類準備，拉取腳本之後的回調函數

初始化操作

調用OpenApi拉取所有腳本

内部連接配接MaxCompute拉取所有DDL腳本内容

測試代碼

測試結果

項目位址

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method