[自研开源] MyData 数据集成之数据过滤 v0.7.2

news2024/11/24 6:55:07

开源地址：gitee | github
详细介绍：MyData 基于 Web API 的数据集成平台
部署文档：用 Docker 部署 MyData
使用手册：MyData 使用手册
试用体验：https://demo.mydata.work
交流Q群：430089673

概述

本篇基于 数据集成之任务流程 介绍任务执行时的数据过滤的使用场景和配置操作。

使用场景

业务系统与mydata集成时，核心是数据的来和去，在这两个方向上分别实现：数据预清洗、数据权限控制；

  1. 数据预清洗，从api获取数据后 过滤排除掉“脏”数据，然后再入库用于数据集成；

    在这里插入图片描述

    例如：接口返回的某字段值不能为空、字段值长度在指定范围等；

    以下代码是 提供数据 类型的任务执行过程：

    // 提供数据
case MdConstant.DATA_PRODUCER:
    // 调用api 获取json
    String json = ApiUtil.read(taskInfo);
    // 将json按字段映射 解析为业务数据
    jobDataService.parseData(taskInfo, json);
    // 根据条件过滤数据
    jobDataFilterService.doFilter(taskInfo);
    // 保存业务数据
    jobDataService.saveTaskData(taskInfo);
    // 更新环境变量
    jobVarService.saveVarValue(taskInfo, json);
	break;

    jobDataFilterService.doFilter 是对数据的预过滤处理，详见 JobDataFilterService.java

    public void doFilter(TaskInfo task) {
    Assert.notNull(task);
	// 获取业务数据
    List<Map> dataList = task.getProduceDataList();
    // 获取配置的过滤条件
    List<BizDataFilter> dataFilters = task.getDataFilters();

    if (CollUtil.isEmpty(dataList) || CollUtil.isEmpty(dataFilters)) {
        return;
    }

    // 定义新的数据集合，用于存储 过滤后的数据
    List<Map> filterDatas = ListUtil.toList();
    // 遍历数据，并进行过滤
    dataList.forEach(data -> {

        boolean isCorrect = false;

        for (BizDataFilter filter : dataFilters) {
            String key = filter.getKey();
            Object filterValue = filter.getValue();
            String op = filter.getOp();

            // 当数据中 不包含 过滤的字段名，则执行下一项过滤
            if (!data.containsKey(key)) {
                continue;
            }

            // 当数据中 指定字段的值 无效，则过滤该数据
            Object dataValue = data.get(key);
            if (ObjectUtil.isNull(dataValue)) {
                isCorrect = true;
                break;
            }

            // 判断业务数据值 和 过滤数据值 都可对比，否则过滤条件无效
            if (!(dataValue instanceof Comparable && filterValue instanceof Comparable)) {
                break;
            }

            String cDataValue = dataValue.toString();
            String cFilterValue = filterValue.toString();
            // 根据op类型，过滤数据
            switch (op) {
                case MdConstant.DATA_OP_EQ:
                    // 等于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) == 0);
                    break;
                case MdConstant.DATA_OP_NE:
                    // 不等于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) != 0);
                    break;
                case MdConstant.DATA_OP_GT:
                    // 大于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) > 0);
                    break;
                case MdConstant.DATA_OP_GTE:
                    // 大于等于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) >= 0);
                    break;
                case MdConstant.DATA_OP_LT:
                    // 小于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) < 0);
                    break;
                case MdConstant.DATA_OP_LTE:
                    // 小于等于
                    isCorrect = (ObjectUtil.compare(cDataValue, cFilterValue) <= 0);
                    break;

                default:
                    throw new RuntimeException("JobDataFilter: 不支持的过滤操作");
            }
        }

        // 当 未被过滤，则添加到过滤结果
        if (isCorrect) {
            filterDatas.add(data);
        }
    });

    task.setProduceDataList(filterDatas);

    task.appendLog("过滤前的业务数据：{}", dataList);
    task.appendLog("过滤条件：{}", dataFilters);
    task.appendLog("过滤后的业务数据：{}", filterDatas);
}

    注：目前0.7版本暂时实现了关系运算，后续增加函数处理；

  2. 数据权限控制，限制应用接收的数据范围，即符合条件的数据才能共享给应用；

    在这里插入图片描述

    以下代码是 消费数据 类型任务的执行过程：

    // 消费数据
case MdConstant.DATA_CONSUMER:
    List<BizDataFilter> filters = taskInfo.getDataFilters();
    if (CollUtil.isNotEmpty(filters)) {
        // 解析过滤条件值中的 自定义字符串
        parseFilterValue(filters);
        // 排除值为null的条件
        filters = filters.stream().filter(filter -> filter.getValue() != null).collect(Collectors.toList());
    }
    String dataCode = taskInfo.getDataCode();
    if (StrUtil.isNotEmpty(dataCode)) {
        // 根据过滤条件 查询数据
        List<Map> dataList = bizDataDAO.list(MdUtil.getBizDbCode(taskInfo.getTenantId(), taskInfo.getProjectId(), taskInfo.getEnvId()), dataCode, filters);
        taskInfo.setConsumeDataList(dataList);
        // 根据字段映射转换为api参数
        jobDataService.convertData(taskInfo);
    }
    // 调用api传输数据
    ApiUtil.write(taskInfo);
    break;

    bizDataDAO.list 方法支持按配置条件查询数据，详见 BizDataDAO.java

    public List<Map> list(String dbCode, String dataCode, List<BizDataFilter> bizDataFilters) {
        MongoTemplate mongoTemplate = mongoFactory.getTemplate(dbCode);
        Query query = new Query();
        // 遍历数据过滤条件
        if (CollUtil.isNotEmpty(bizDataFilters)) {
            // mongodb的查询条件集合
            List<Criteria> criteriaList = CollUtil.newArrayList();
            for (BizDataFilter bizDataFilter : bizDataFilters) {
                // 条件key
                String key = bizDataFilter.getKey();
                // 条件操作
                String op = bizDataFilter.getOp();
                // 条件值
                Object value = bizDataFilter.getValue();

                // 根据条件操作类型 调用mongodb对应的查询方法
                Criteria criteria = Criteria.where(key);
                switch (op) {
                    case MdConstant.DATA_OP_EQ:
                        criteria.is(value);
                        break;
                    case MdConstant.DATA_OP_NE:
                        criteria.ne(value);
                        break;
                    case MdConstant.DATA_OP_GT:
                        criteria.gt(value);
                        break;
                    case MdConstant.DATA_OP_GTE:
                        criteria.gte(value);
                        break;
                    case MdConstant.DATA_OP_LT:
                        criteria.lt(value);
                        break;
                    case MdConstant.DATA_OP_LTE:
                        criteria.lte(value);
                        break;

                    default:
                        throw new RuntimeException("BizDataDAO: 不支持的过滤操作");
                }
                // 存入mongodb的查询条件集合
                criteriaList.add(criteria);
            }

            // mongodb查询条件集合 加入查询中
            query.addCriteria(new Criteria().andOperator(criteriaList));
        }

        // 执行查询
        return mongoTemplate.find(query, Map.class, dataCode);
    }

配置操作

  1. 创建任务过程请参考 使用手册

  2. 在创建任务界面中，添加数据过滤条件

    如下图过滤条件是 salary > 600
    在这里插入图片描述

  3. 执行任务后 通过日志详情可以看到数据入库前预清洗；
    在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1522624.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Ubuntu 虚拟机安装

Ubuntu 虚拟机安装

最小化安装后常用工具 sudo apt-get install vim# ifconfig apt install net-tools # nload apt install nload # 很多都要用到 apt install build-essential # 开发相关 apt install gcc gapt install iproute2 ntpdate tcpdump telnet traceroute \ nfs-kernel-server nfs…
阅读更多...
Java项目：57 ssm011线上旅行信息管理系统ssm+vue

Java项目：57 ssm011线上旅行信息管理系统ssm+vue

作者主页&#xff1a;源码空间codegym 简介&#xff1a;Java领域优质创作者、Java项目、学习资料、技术互助 文中获取源码 项目介绍 本线上旅行信息管理系统&#xff0c;主要实现了用户功能模块和管理员功能模块两大部分 用户可查看旅行相关信息&#xff0c;注册登录后还可实…
阅读更多...
第十二届蓝桥杯EDA省赛真题分析

第十二届蓝桥杯EDA省赛真题分析

前言&#xff1a; 第十二届蓝桥杯EDA比赛用的是AD软件&#xff0c;从第十四届起都是使用嘉立创EDA专业版&#xff0c;所以在这里我用嘉立创EDA专业版实现题目要求。 一、省赛第一套真题题目 主观题整套题目如下&#xff1a; 试题一&#xff1a;库文件设计&#xff08;5分&am…
阅读更多...
Django 解决新建表删除后无法重新创建等问题

Django 解决新建表删除后无法重新创建等问题

Django 解决新建表删除后无法重新创建等问题 问题发生描述处理办法首先删除了app对应目录migrations下除 __init__.py以外的所有文件:然后&#xff0c;删除migrations中关于你的app的同步数据数据库记录最后&#xff0c;重新执行迁移插入 问题发生描述 Django创建的表&#xf…
阅读更多...
【计算机视觉】二、图像形成——实验：2D变换编辑（Pygame）

【计算机视觉】二、图像形成——实验：2D变换编辑（Pygame）

文章目录 一、向量和矩阵的基本运算二、几何基元和变换1、几何基元(Geometric Primitives)2、几何变换(Geometric Transformations)2D变换编辑器0. 程序简介环境说明程序流程 1. 各种变换平移变换旋转变换等比缩放变换缩放变换镜像变换剪切变换 2. 按钮按钮类创建按钮 3. Pygam…
阅读更多...
前端vue-Taro框架中使用插件 ---pinyin 将城市树形分类

前端vue-Taro框架中使用插件 ---pinyin 将城市树形分类

1.需求 当我做一个获取城市的功能的时候 我发向后端返回的数据 和我想i选要的相差太多 这样的在手机端可以滑动 并且 快捷选中的城市列表 目前的数据是这样的&#xff0c;就是一个城市数组 目前这样的数组 我要想显示我的页面实现功能是不行的 需要是树形结够 所以我前端…
阅读更多...
CI/CD实战-git工具使用 1

CI/CD实战-git工具使用 1

版本控制系统 本地版本控制系统 集中化的版本控制系统 分布式版本控制系统 git官网文档&#xff1a;https://git-scm.com/book/zh/v2 Git 有三种状态&#xff1a;已提交&#xff08;committed&#xff09;、已修改&#xff08;modified&#xff09; 和 已暂存&#xff08;sta…
阅读更多...
【CTF web1】

【CTF web1】

CTF web 一、CTF web -PHP弱类型1、是否相等&#xff1f;2、转换规则: 二、CTF web -md5绕过1、若类型比较绕过2、null绕过3、碰撞绕过 三、习题 一、CTF web -PHP弱类型 1、是否相等&#xff1f; &#xff1a;在进行比较的时候&#xff0c;会先判断两种字符串的类型是否相等&…
阅读更多...
EVENG环境安装及测试 1

EVENG环境安装及测试 1

文章目录 下载eve镜像导入镜像访问测试导入自定义镜像 下载eve镜像 下载地址 链接&#xff1a;https://pan.baidu.com/s/1NqGE34oE5qZ6TCugMymPDg 提取码&#xff1a;f4m1 导入镜像 安装vmware 虚拟机&#xff0c;文件->打开 选中上述镜像 输入虚拟机的名称和保存 路径&a…
阅读更多...
接口幂等性问题和常见解决方案

接口幂等性问题和常见解决方案

接口幂等性问题和常见解决方案 1.什么是接口幂等性问题1.1 会产生接口幂等性的问题1.2 解决思路 2.接口幂等性的解决方案2.1 唯一索引解决方案2.2 乐观锁解决方案2.3 分布式锁解决方案2.4 Token解决方案(最优方案) 1.什么是接口幂等性问题 幂等性: 用户同一操作发起的一次或多…
阅读更多...
无人机助力智慧农田除草新模式，基于YOLOv5全系列【n/s/m/l/x】参数模型开发构建无人机航拍场景下的农田杂草检测识别系统

无人机助力智慧农田除草新模式，基于YOLOv5全系列【n/s/m/l/x】参数模型开发构建无人机航拍场景下的农田杂草检测识别系统

科技发展到今天&#xff0c;无人机喷洒药物已经不是一件新鲜事情了&#xff0c;在很多高危的工作领域中&#xff0c;比如高空电力设备除冰&#xff0c;电力设备部件传送更换等等&#xff0c;无人机都可以扮演非常出色的作用&#xff0c;前面回到老家一段时间&#xff0c;最近正…
阅读更多...
Linux系统——Session ID（负载均衡如何保持会话）

Linux系统——Session ID（负载均衡如何保持会话）

目录 一、实验环境搭建 二、部署Nginx代理服务器配置 三、部署后端真是服务器Tomcat配置 四、配置Tomcat的Session ID会话保持 五、测试 此次实验是Tomcat后端服务器如何做Session ID会话保持 一、实验环境搭建 [rootlocalhost ~]#systemctl stop firewalld [rootlocalho…
阅读更多...
CodeTop day3

CodeTop day3

class Solution {public int[] sortArray(int[] nums) {//这种方法超时【快速排序】for (int i0;i<nums.length-1;i){int minIndex i;//假设当前开始下标为最小元素下标for (int ji1;j<nums.length;j){//从i到nums。length-1区间里找到最小值下标if (nums[j]<nums[mi…
阅读更多...
腾讯云图形验证码的PHP示例

腾讯云图形验证码的PHP示例

需要准备的 1.API密钥 SecretId 及 SecretKey 两部分&#xff0c; SecretId 用于标识 API 调用者的身份&#xff0c; SecretKey 用于加密签名字符串和服务器端验证签名字符串的密钥。 前往API密钥管理页面&#xff0c;即可进行获取 https://console.cloud.tencent.com/cam/ca…
阅读更多...
鸿蒙Harmony应用开发—ArkTS声明式开发（容器组件：UIExtensionComponent (系统接口)）

鸿蒙Harmony应用开发—ArkTS声明式开发（容器组件：UIExtensionComponent (系统接口)）

UIExtensionComponent用于支持在本页面内嵌入其他应用提供的UI。展示的内容在另外一个进程中运行&#xff0c;本应用并不参与其中的布局和渲染。 通常用于有进程隔离诉求的模块化开发场景。 说明&#xff1a; 该组件从API Version 10开始支持。后续版本如有新增内容&#xff0…
阅读更多...
Linux服务器（RedHat、CentOS系）安全相关巡检shell脚本

Linux服务器（RedHat、CentOS系）安全相关巡检shell脚本

提示&#xff1a;巡检脚本可以使用crontab定时执行&#xff0c;人工根据执行结束时间点统一收集报告文件即可。 #!/bin/bash# Define output file current_date$(date "%Y%m%d") # Gets the current date in YYYYMMDD format echo >server_security_inspection_r…
阅读更多...
Python基础综合案例-数据可视化

Python基础综合案例-数据可视化

一、数据可视化 - 折线图可视化 1.1、json数据格式 """ 演示JSON数据和Python字典的相互转换 """ import json # 准备列表&#xff0c;列表内每一个元素都是字典&#xff0c;将其转换为JSON data [{"name": "张大山", &quo…
阅读更多...
鸿蒙Harmony应用开发—ArkTS声明式开发（容器组件：Stack）

鸿蒙Harmony应用开发—ArkTS声明式开发（容器组件：Stack）

堆叠容器&#xff0c;子组件按照顺序依次入栈&#xff0c;后一个子组件覆盖前一个子组件。 说明&#xff1a; 该组件从API Version 7开始支持。后续版本如有新增内容&#xff0c;则采用上角标单独标记该内容的起始版本。 子组件 可以包含子组件。 接口 Stack(value?: { ali…
阅读更多...
深度学习-基于机器学习的情绪分析研究

深度学习-基于机器学习的情绪分析研究

概要 互联网技术的迅速发展使得社交平台逐渐成为热点事件中社会情感的枢纽。社会热点事件的舆论监管的其中一个重要环节就是能够准确分析民众的社会情绪。本文旨在探索可以基于文本大数据彻底分析民众对热点事件的社会情绪的模型和方法。先是从社交平台上借助文本大数据、对数据…
阅读更多...
实战：django项目环境搭建(pycharm，virtualBox)

实战：django项目环境搭建(pycharm，virtualBox)

django项目环境搭建 一.创建虚拟环境二.创建PyCharm远程连接 一.创建虚拟环境 需要用到的软件&#xff1a;PyCharm&#xff0c;VirtualBox虚拟机。 1.打开虚拟机终端&#xff0c;创建新的虚拟环境 Book。 2.在虚拟环境中创建新的文件夹 library&#xff0c;cd命令进入该文件…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023