data gateway_交通运输部监测调度

全栈程序员站长

发布于 2022-11-16 18:31:29

5390

发布于 2022-11-16 18:31:29

对于大数据场景，计算的业务逻辑基本都在数据处理过程中完成，最后往往会持久化下来计算的结果，存储在mysql，es，hbase等适合提供查询的地方。而对于在不同数据库上的数据，需要开发接口，管理接口也是一个不小的工作量。Dataway的出现，正好解决了这个场景下的接口开发，管理等问题。只需要通过SQL，Dataql等配置就能完成一个接口的上线，大大提高了开发效率，以及管理成本。对于数据中台建设中，Dataway可以使用在统一服务层，对接各种数据源，管理数据出口。

一、安装

引入依赖

Dataway 是 Hasor 生态中的一员，使用 Dataway 第一步需要通过 hasor-spring 打通两个生态。

<!-- 引入依赖 -->
<dependency>
    <groupId>net.hasor</groupId>
    <artifactId>hasor-spring</artifactId>
    <version>4.2.2</version>
    <!-- 查看最新版本：https://mvnrepository.com/artifact/net.hasor/hasor-spring -->
</dependency>
<dependency>
    <groupId>net.hasor</groupId>
    <artifactId>hasor-dataway</artifactId>
    <version>4.2.2</version>
    <!-- 查看最新版本：https://mvnrepository.com/artifact/net.hasor/hasor-dataway -->
</dependency>

Jetbrains全家桶1年46，售后保障稳定

启用 Hasor

import net.hasor.spring.boot.EnableHasor;
import net.hasor.spring.boot.EnableHasorWeb;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

import static net.hasor.spring.boot.WorkAt.Interceptor;


@EnableHasor
@EnableHasorWeb(at =Interceptor )
@SpringBootApplication
@Configuration
public class DatawayApplication implements WebMvcConfigurer {

	public static void main(String[] args) {
		SpringApplication.run(DatawayApplication.class, args);
	}

}

启用 Dataway

然后第二步，在应用的 application.properties 配置文件中启用 Dataway

# 启用 Dataway 功能（默认不启用）
HASOR_DATAQL_DATAWAY=true
# 开启 ui 管理功能（注意生产环境必须要设置为 false，否则会造成严重的生产安全事故）
HASOR_DATAQL_DATAWAY_ADMIN=true
 
# （可选）API工作路径
HASOR_DATAQL_DATAWAY_API_URL=/api/
# （可选）ui 的工作路径，只有开启 ui 管理功能后才有效
HASOR_DATAQL_DATAWAY_UI_URL=/interface-ui/
# SQL执行器方言设置（可选，建议设置）
HASOR_DATAQL_FX_PAGE_DIALECT=mysql

初始化必要的表(例：MySQL)

CREATE TABLE interface_info (
  api_id          varchar(64)  NOT NULL COMMENT 'ID',
  api_method      varchar(12)  NOT NULL COMMENT 'HttpMethod：GET、PUT、POST',
  api_path        varchar(512) NOT NULL COMMENT '拦截路径',
  api_status      varchar(4)   NOT NULL COMMENT '状态：-1-删除, 0-草稿，1-发布，2-有变更，3-禁用',
  api_comment     varchar(255) NOT NULL COMMENT '注释',
  api_type        varchar(24)  NOT NULL COMMENT '脚本类型：SQL、DataQL',
  api_script      mediumtext   NOT NULL COMMENT '查询脚本：xxxxxxx',
  api_schema      mediumtext   NOT NULL COMMENT '接口的请求/响应数据结构',
  api_sample      mediumtext   NOT NULL COMMENT '请求/响应/请求头样本数据',
  api_option      mediumtext   NOT NULL COMMENT '扩展配置信息',
  api_create_time varchar(32)  NOT NULL COMMENT '创建时间',
  api_gmt_time    varchar(32)  NOT NULL COMMENT '修改时间',
  PRIMARY KEY (api_id),
  UNIQUE KEY uk_interface_info (api_path)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Dataway 中的API';
 
CREATE TABLE interface_release (
  pub_id           varchar(64)  NOT NULL COMMENT 'Publish ID',
  pub_api_id       varchar(64)  NOT NULL COMMENT '所属API ID',
  pub_method       varchar(12)  NOT NULL COMMENT 'HttpMethod：GET、PUT、POST',
  pub_path         varchar(512) NOT NULL COMMENT '拦截路径',
  pub_status       varchar(4)   NOT NULL COMMENT '状态：-1-删除, 0-草稿，1-发布，2-有变更，3-禁用',
  pub_comment      varchar(255) NOT NULL COMMENT '注释',
  pub_type         varchar(24)  NOT NULL COMMENT '脚本类型：SQL、DataQL',
  pub_script       mediumtext   NOT NULL COMMENT '查询脚本：xxxxxxx',
  pub_script_ori   mediumtext   NOT NULL COMMENT '原始查询脚本，仅当类型为SQL时不同',
  pub_schema       mediumtext   NOT NULL COMMENT '接口的请求/响应数据结构',
  pub_sample       mediumtext   NOT NULL COMMENT '请求/响应/请求头样本数据',
  pub_option       mediumtext   NOT NULL COMMENT '扩展配置信息',
  pub_release_time varchar(32)  NOT NULL COMMENT '发布时间（下线不更新）',
  PRIMARY KEY (pub_id),
  KEY idx_interface_release_api  (pub_api_id),
  KEY idx_interface_release_path (pub_path)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Dataway API 发布历史。'

初始化数据源

@DimModule
@Component
public class ExampleModule implements SpringModule {
    @Autowired
    private DataSource dataSource = null;
 
    public void loadModule(ApiBinder apiBinder) throws Throwable {
        // .DataSource form Spring boot into Hasor
        apiBinder.installModule(new JdbcModule(Level.Full, this.dataSource));
        // .custom DataQL
        //apiBinder.tryCast(QueryApiBinder.class).loadUdfSource(apiBinder.findClass(DimUdfSource.class));
        //apiBinder.tryCast(QueryApiBinder.class).bindFragment("sql", SqlFragment.class);
    }
}

最后一步，将 Spring 使用的数据源导入到 Hasor 环境共 Dataway 使用。

启动工程

2020-04-01 09:13:18.502 [main] INFO  n.h.core.context.TemplateAppContext - loadModule class net.hasor.dataway.config.DatawayModule
2020-04-01 09:13:18.502 [main] INFO  n.hasor.dataway.config.DatawayModule - dataway api workAt /api/
2020-04-01 09:13:18.502 [main] INFO  n.h.c.e.AbstractEnvironment - var -> HASOR_DATAQL_DATAWAY_API_URL = /api/.
2020-04-01 09:13:18.515 [main] INFO  n.hasor.dataway.config.DatawayModule - dataway admin workAt /interface-ui/

– dataway api workAt /api/ 表示 API 的工作路径。 – dataway admin workAt /interface-ui/ 表示管理配置界面的地址。

此时访问：http://<yourIP>:<yourProt>/interface-ui/ 就可以看到配置页面了。

高可用

多节点部署，通过nginx等做负载均衡

二、操作界面

新增

点击页面顶部的 New 就可以进入新增接口页面。

在新增页面中，最主要的是要设置API的请求地址和方法。在保存 API 的时候 dataway 会对 Api 地址进行校验。

合法的API地址中只能包含下列字符： ! $ ‘ ( ) * + , – . / : ; = @ _ ~ [0-9] [a-z] [A-Z]

编辑器中开放了 POST、PUT、GET、DELETE 四个可选的 Http Method。

在保存时如果遇到错误会有类似下面这样的提示（下列是，API地址冲突的错误提示）

提示

请注意 API 创建就不允许修改请求地址和方法，可以删除重建API来达到更名目的。

新增接口时会展示一段 HelloWord 用例，这个用例中包含了请求一个接口传入一个 message 参数，然后利用 DataQL 查询获取这个参数并将其返回。

开发者可以在新建接口时直接点击工具条中的 Execute 图标执行得到运行结果。

执行

在任意的编辑器界面中（新增模式 or 编辑模式），都可以直接在编辑区编写 DataQL 查询并通过右上角的 Execute 按钮执行。

发布

当接口开发配置完成，需要将其发布以供使用。一个接口的发布上线要经历三个过程，具体如下：

Execute

调试当前编辑器中的 DataQL 查询。

Smork Test

冒烟测试。和 Execute 不同，Smork 同样是执行 DataQL 查询并要求查询正常执行完毕。但是区别 Execute 的是 Smork 不会使用编辑器中的 DataQL 查询语句，而是到数据库中获取对应的查询语句。

因为接口的发布也是将数据中的 DataQL 查询语句进行发布。

Publish

当冒烟测试通过之后就可以点击发布按钮把接口发布上线了。每次发布 Dataway 都会在 interface_release 表中新增一条记录。

删除/下线

已经发布上线的接口只能执行接口下线操作。

已经下线的接口或者正在编辑中的接口可以执行删除操作，删除操作会物理删除 interface_info 表中对应的记录。

但是删除操作并不会删除曾经的发布历史，这就留给我们了一条可以找回被删除的接口曾经发布过的历史记录。只不过这一步只能进入数据库中自行搜索。

发布历史

每个 Dataway 上的接口在发布时都会在 interface_release 表中生成一条记录。发布历史看的就是这里的记录。

在历史记录列表的右侧 icon 按可以恢复历史记录的内容到编辑器中。

三、ES支持

为了方便对ES的访问，增加了Dataway对Es的Sql支持；

引入依赖

<dependency>
            <groupId>org.es</groupId>
            <artifactId>sql2dsl-toolkit</artifactId>
            <version>5.4.0</version>
</dependency>

由于我们的es5.4，版本比较低，所以引入的sql2dsl-toolkit

SQL语法参见： https://github.com/gitchennan/elasticsearch-query-toolkit/wiki/elasticsearch-query-toolkit-help-doc

添加ES属性到配置文件 application.properties

spring.data.elasticsearch.cluster-nodes=host1:port1,host2:port2
spring.data.elasticsearch.cluster-name=xxxx

增加ESSQL执行器

执行器是 DataQL 的一个 FragmentProcess 扩展

import net.hasor.core.DimModule;
import net.hasor.dataql.DimFragment;
import net.hasor.dataql.FragmentProcess;
import net.hasor.dataql.Hints;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.search.SearchHit;
import org.es.sql.bean.ElasticSqlParseResult;
import org.es.sql.parser.ElasticSql2DslParser;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

import static java.util.Arrays.stream;

@DimModule
@Component
@DimFragment("esV2")
public class ESQueryV2 implements FragmentProcess {

    @Autowired
    private Client esClient;


    @Override
    public Object runFragment(Hints hint, Map<String, Object> params, String fragmentString) throws Throwable {

//        System.out.println("参数："+params);
        /**
         * sql 语法见 ：https://github.com/gitchennan/elasticsearch-query-toolkit/wiki/elasticsearch-query-toolkit-help-doc
         */

        //获取 fragmentString 中#{xx} 字符替换
        String runsql = fragmentString;
        for(String key:params.keySet()){
            runsql = runsql.replace("#{"+key+"}",params.get(key).toString());
        }
        System.out.println(runsql);

        ElasticSql2DslParser sql2DslParser = new ElasticSql2DslParser(); //解析SQL
        ElasticSqlParseResult parseResult = sql2DslParser.parse(runsql); //生成DSL(可用于rest api调用)
        String dsl = parseResult.toDsl();
        System.out.println(dsl);


        //Client esClient = init();
        //toRequest方法接收一个clinet对象参数
        SearchRequestBuilder searchReq = parseResult.toRequest(esClient); //执行查询
        SearchResponse response = searchReq.execute().actionGet();
        SearchHit[] searchHits  = response.getHits().getHits();

        return stream(searchHits).map(x -> x.getSource()).toArray();
    }
}

新建HasorModule配置

import net.hasor.core.DimModule;
import net.hasor.dataql.QueryApiBinder;
import net.hasor.dataql.QueryModule;
import net.hasor.db.JdbcModule;
import net.hasor.db.Level;
import net.hasor.spring.SpringModule;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;

//import net.hasor.dataql.fx.db.LookupDataSourceListener;


@DimModule
@Component
public class DatawayModule implements SpringModule,QueryModule {

    @Autowired
    private DataSource dataSource = null;


    @Override
    public void loadModule(QueryApiBinder apiBinder) throws Throwable {
        apiBinder.installModule(new JdbcModule(Level.Full, this.dataSource));
        apiBinder.loadFragment(ESQueryV2.class,springTypeSupplier(apiBinder));  //集成sql2dsl-toolkit

    }
}

重新启动服务即可

效果