记csv、parquet数据预览一个bug的解决

2024-01-14 06:12

本文主要是介绍记csv、parquet数据预览一个bug的解决,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 一、概述
  • 二、实现过程
    • 1. 业务流程如图:
    • 2. 业务逻辑
    • 3. 运行结果
  • 三、bug现象
    • 1. 单元测试
    • 2.运行结果
  • 三、流程梳理
    • 1. 方向一
    • 2. 方向二

一、概述

工作中遇到通过sparksession解析csv、parquet文件并预览top100的需求。

二、实现过程

1. 业务流程如图:

hiveSQL读取数据
数据写入csv或parquet文件
预览csv或parquet文件top100数据

2. 业务逻辑

为了便于测试,我们下面以单元测试中模拟数据来说明


import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;import com.alibaba.fastjson.JSONObject;import lombok.extern.slf4j.Slf4j;@Slf4j
public class GroupingByDataTest
{static List<String> result = new ArrayList<>();@BeforeAllpublic static void init(){result.add("{\"student_no\":\"0204006\",\"student_name\":\"学生6\",\"field\":\"项目6\",\"value2\":\"6\",\"sex\":\"女\"}");result.add("{\"student_no\":\"0204006\",\"student_name\":\"学生6\",\"field\":\"项目6\",\"value2\":\"6\",\"sex\":\"女\"}");result.add("{\"student_no\":\"0204006\",\"student_name\":\"学生6\",\"field\":\"项目6\",\"value2\":\"6\",\"sex\":\"女\"}");result.add("{\"student_no\":\"0204006\",\"student_name\":\"学生6\",\"field\":\"项目6\",\"value2\":\"6\",\"sex\":\"女\"}");}@Testpublic void test002(){Map<Object, List<Object>> r = result.stream().map(s -> JSONObject.parseObject(s).entrySet()) // map.flatMap(m -> m.stream()) // flatMap.collect(Collectors.groupingBy(mp -> mp.getKey(), Collectors.mapping(x -> x.getValue(), Collectors.toList())));log.info("{}", r);}
}

3. 运行结果

 com.fly.lambda.GroupingByDataTest - {student_name=[学生6, 学生6, 学生6, 学生6], student_no=[0204006, 0204006, 0204006, 0204006], value2=[6, 6, 6, 6], field=[项目6, 项目6, 项目6, 项目6], sex=[女, 女, 女, 女]}

目前看来一切正常。

三、bug现象

实际测试过程中发现,hive数据仓库中的字段由于各种原因并不一定都有值,从而导致csv、parquet保存结果时字段为空

1. 单元测试


import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;import com.alibaba.fastjson.JSONObject;import lombok.extern.slf4j.Slf4j;@Slf4j
public class GroupingByDataTest
{static List<String> result = new ArrayList<>();@BeforeAllpublic static void init(){result.add("{\"student_name\":\"学生1\",\"student_no\":\"0204001\",\"field\":\"项目1\",                 \"sex\":\"男\"}");result.add("{\"student_name\":\"学生2\",\"student_no\":\"0204002\",\"field\":\"项目2\",\"value2\":\"2\"               }");result.add("{\"student_name\":\"学生3\",                           \"field\":\"项目3\",\"value2\":\"3\",\"sex\":\"女\"}");result.add("{                           \"student_no\":\"0204004\",\"field\":\"项目4\",\"value2\":\"4\",\"sex\":\"男\"}");result.add("{\"student_name\":\"学生5\",\"student_no\":\"0204005\",\"field\":\"项目5\",\"value2\":\"5\",\"sex\":\"女\"}");result.add("{\"student_no\":\"0204006\",\"student_name\":\"学生6\",\"field\":\"项目6\",\"value2\":\"6\",\"sex\":\"女\"}");}@Testpublic void test002(){Map<Object, List<Object>> r = result.stream().map(s -> JSONObject.parseObject(s).entrySet()) // map.flatMap(m -> m.stream()) // flatMap.collect(Collectors.groupingBy(mp -> mp.getKey(), Collectors.mapping(x -> x.getValue(), Collectors.toList())));log.info("{}", r);}
}

2.运行结果

 com.fly.lambda.GroupingByDataTest - {student_name=[学生1, 学生2, 学生3, 学生5, 学生6], student_no=[0204001, 0204002, 0204004, 0204005, 0204006], value2=[2, 3, 4, 5, 6], field=[项目1, 项目2, 项目3, 项目4, 项目5, 项目6], sex=[男, 女, 男, 女, 女]}

期望的结果为

 com.fly.lambda.GroupingByDataTest - before : {student_name=[学生1, 学生2, 学生3, null, 学生5, 学生6], student_no=[0204001, 0204002, null, 0204004, 0204005, 0204006], value2=[null, 2, 3, 4, 5, 6], field=[项目1, 项目2, 项目3, 项目4, 项目5, 项目6], sex=[男, null, 女, 男, 女, 女]}

三、流程梳理

解决这个问题有2个方向

1. 方向一

从数据来源解决,也就是 hiveSQL读取数据使用 coalsce 函数进行空值处理,实际去解决的过程中发现2个问题。

  1. 强制业务用户编辑hiveSQL时显式调用(用户体验太差,增加使用难度
  2. 不强制业务用户编辑hiveSQL时显式调用,后台接受到SQL后自动添加coalsce 函数(后台业务逻辑复杂,eg: 使用了条件语句、多表关联查询等等情况。不一而足,几乎没法妥善处理

2. 方向二

hiveSQL读取数据
数据写入csv或parquet文件
预览csv或parquet文件top100数据

hiveSQL读取数据、数据写入csv或parquet文件正常进行,不用特殊处理, 修改步骤3

分为2步骤,步骤1,遍历获取全部的key去重,步骤2,自动对缺失数据的key补充空值

核心代码如下:

@Testpublic void test003()throws IOException{// 取keysList<String> keys = result.stream().map(s -> JSONObject.parseObject(s).entrySet()).flatMap(m -> m.stream()).map(r -> r.getKey()).distinct().collect(Collectors.toList());keys.stream().forEach(log::info);Map<String, List<Object>> r = result.stream().map(s -> parse(s, keys)).flatMap(m -> m.stream()) // flatMap.collect(Collectors.groupingBy(mp -> mp.getKey(), Collectors.mapping(x -> x.getValue(), Collectors.toList())));log.info("before : {}", r);log.info("sorted : {}", new TreeMap<>(r));}/*** 设置value, 根据需要补充空值*/private Set<Entry<String, Object>> parse(String s, List<String> keys){JSONObject jsonObject = JSONObject.parseObject(s);keys.stream().forEach(key -> {if (!jsonObject.containsKey(key)){jsonObject.put(key, null);}});return jsonObject.entrySet();}

可以说,花比较小的成本,以比较少的代码变动,相对稳妥的解决了问题。


有任何问题和建议,都可以向我提问讨论,大家一起进步,谢谢!

-over-

这篇关于记csv、parquet数据预览一个bug的解决的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/604177

相关文章

Python实现批量CSV转Excel的高性能处理方案

《Python实现批量CSV转Excel的高性能处理方案》在日常办公中,我们经常需要将CSV格式的数据转换为Excel文件,本文将介绍一个基于Python的高性能解决方案,感兴趣的小伙伴可以跟随小编一... 目录一、场景需求二、技术方案三、核心代码四、批量处理方案五、性能优化六、使用示例完整代码七、小结一、

MyBatis-plus处理存储json数据过程

《MyBatis-plus处理存储json数据过程》文章介绍MyBatis-Plus3.4.21处理对象与集合的差异:对象可用内置Handler配合autoResultMap,集合需自定义处理器继承F... 目录1、如果是对象2、如果需要转换的是List集合总结对象和集合分两种情况处理,目前我用的MP的版本

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映

504 Gateway Timeout网关超时的根源及完美解决方法

《504GatewayTimeout网关超时的根源及完美解决方法》在日常开发和运维过程中,504GatewayTimeout错误是常见的网络问题之一,尤其是在使用反向代理(如Nginx)或... 目录引言为什么会出现 504 错误?1. 探索 504 Gateway Timeout 错误的根源 1.1 后端

解决升级JDK报错:module java.base does not“opens java.lang.reflect“to unnamed module问题

《解决升级JDK报错:modulejava.basedoesnot“opensjava.lang.reflect“tounnamedmodule问题》SpringBoot启动错误源于Jav... 目录问题描述原因分析解决方案总结问题描述启动sprintboot时报以下错误原因分析编程异js常是由Ja

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章:LiteDB时间序列数据模型设计1.1 核心设计原则

Java+AI驱动实现PDF文件数据提取与解析

《Java+AI驱动实现PDF文件数据提取与解析》本文将和大家分享一套基于AI的体检报告智能评估方案,详细介绍从PDF上传、内容提取到AI分析、数据存储的全流程自动化实现方法,感兴趣的可以了解下... 目录一、核心流程:从上传到评估的完整链路二、第一步:解析 PDF,提取体检报告内容1. 引入依赖2. 封装

深度剖析SpringBoot日志性能提升的原因与解决

《深度剖析SpringBoot日志性能提升的原因与解决》日志记录本该是辅助工具,却为何成了性能瓶颈,SpringBoot如何用代码彻底破解日志导致的高延迟问题,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言第一章:日志性能陷阱的底层原理1.1 日志级别的“双刃剑”效应1.2 同步日志的“吞吐量杀手”

MySQL 表空却 ibd 文件过大的问题及解决方法

《MySQL表空却ibd文件过大的问题及解决方法》本文给大家介绍MySQL表空却ibd文件过大的问题及解决方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考... 目录一、问题背景:表空却 “吃满” 磁盘的怪事二、问题复现:一步步编程还原异常场景1. 准备测试源表与数据

解决Nginx启动报错Job for nginx.service failed because the control process exited with error code问题

《解决Nginx启动报错Jobfornginx.servicefailedbecausethecontrolprocessexitedwitherrorcode问题》Nginx启... 目录一、报错如下二、解决原因三、解决方式总结一、报错如下Job for nginx.service failed bec