拉链表-spark版本

采用spark实现的拉链表

拉链表初始化

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.lit/*** 拉链表初始化*/
object table_zip_initial {val lastDay = "9999-12-31"def main(args: Array[String]): Unit = {var table_base = "t_uac_organization" //基表var table_zip = "ods_uac_org_zip" //拉链表/*** 基于该天的t_uac_organization*/var dt = "2023-01-31"System.setProperty("HADOOP_USER_NAME", "root")val builder = SparkUtils.getBuilderif (System.getProperties.getProperty("os.name").contains("Windows")) {builder.master("local[*]")} else {table_base = args(0)table_zip = args(1)dt = args(2)}val spark = builder.appName(this.getClass.getName).getOrCreate()val hive_db = "common"spark.sql(s"use $hive_db")/*** 初始化，一次*/if (!TableUtils.tableExists(spark, hive_db, table_zip)) {println(s"$table_zip not exists,初始化")init(dt, spark, hive_db, table_base, table_zip)} else {val t_zip = spark.sql(s"""||select * from $table_zip where dt='$lastDay'||""".stripMargin)if (t_zip.isEmpty) {//initprintln(s"$table_zip isEmpty 初始化")init(dt, spark, hive_db, table_base, table_zip)} else {println(s"$table_zip exist and not empty,无需初始化!!!")}}spark.stop()}private def init(dt: String, spark: SparkSession, hive_db: String, table_base: String, table_zip: String): Unit = {val t_base = spark.sql(s"""||select * from $table_base where dt='${dt}'|""".stripMargin)println(s"$table_base show")t_base.show(false)val ods_zip = t_base.drop("dt").withColumn("t_start", lit(dt)).withColumn("t_end", lit(lastDay)).withColumn("dt", lit(lastDay))if (!ods_zip.isEmpty) {println(s"$table_zip show")ods_zip.show(false)println(s"$table_zip 初始化...")SinkUtil.sink_to_hive(lastDay, spark, ods_zip, hive_db, hive_table = s"$table_zip", "parquet", MySaveMode.OverWriteByDt)} else {println(s"$table_zip is empty,初始化失败...")}}
}

拉链表每日滚动计算

import org.apache.spark.sql.functions.{count, lit}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel/*** 明智优点组织架构拉链表* 记录其状态变化过程* 拉链表只能从装载首日起，一天一天滚动计算*/
object ods_uac_org_zip {val lastDay = "9999-12-31"def main(args: Array[String]): Unit = {var dt = "2023-02-01"var dt1 = "2023-02-02"System.setProperty("HADOOP_USER_NAME", "root")val builder = SparkUtils.getBuilderif (System.getProperties.getProperty("os.name").contains("Windows")) {builder.master("local[*]")} else {dt = args(0)dt1 = args(1)}val spark = builder.appName(this.getClass.getName).getOrCreate()val hive_db = "common"spark.sql(s"use $hive_db")new IDate {override def onDate(dt: String): Unit = {processByDt(spark, dt, hive_db)}}.invoke(dt, dt1)spark.stop()}/*** 滚动计算每个dt的对应的过期数据*/def processByDt(spark: SparkSession, dt: String, hive_db: String): Unit = {val theDayBeforeDt = DateUtil.back1Day(dt + " 00:00:00").split(" ")(0)/*** 一定需要先缓存* 否则重算则fileNotFoundException* 但是缓存也有坑，不一定所有的数据都会缓存* 因此需要借助临时表处理或者设置ck*///TODOvar ods_uac_org_zip = spark.sql(s"""||select * from ods_uac_org_zip where dt='$lastDay'|""".stripMargin).persist(StorageLevel.MEMORY_ONLY_SER_2)/*** 持久化为临时表*/ods_uac_org_zip.repartition(3).write.format("parquet").mode(SaveMode.Overwrite).saveAsTable(s"${hive_db}.ods_uac_org_zip_tmp")/*** 已经指向临时表* 后续方便对源表(ods_uac_org_zip)进行更新*/ods_uac_org_zip = spark.sql(s"""||select * from ods_uac_org_zip_tmp|""".stripMargin)/*** old,已经存在的拉链表的最新全量*/val f_old_9999 = ods_uac_org_zip.drop("dt")println("f_old_9999 show")f_old_9999.show(false)/*** dt该天的新增和变化*/val f_new = spark.sql(s"""||select * from new_change_t_uac_organization where dt='${dt}'|""".stripMargin).drop("dt").withColumnRenamed("id", "id2").withColumnRenamed("org_name", "org_name2").withColumnRenamed("parent_id", "parent_id2").withColumnRenamed("sort", "sort2").withColumnRenamed("org_type", "org_type2").withColumnRenamed("org_level", "org_level2").withColumnRenamed("is_auth_scope", "is_auth_scope2").withColumnRenamed("parent_auth_scope_id", "parent_auth_scope_id2").withColumnRenamed("status", "status2").withColumnRenamed("icon_class", "icon_class2").withColumnRenamed("create_id", "create_id2").withColumnRenamed("create_time", "create_time2").withColumnRenamed("update_id", "update_id2").withColumnRenamed("update_time", "update_time2").withColumnRenamed("version", "version2").withColumn("t_start2", lit(dt)).withColumn("t_end2", lit(lastDay))println("f_new show")f_new.show(false)val f1 = f_old_9999.join(f_new, f_old_9999.col("id") === f_new.col("id2"), "full_outer")f1.createOrReplaceTempView("v1")println("v1 temp show")f1.show(false)f1.filter(s"id='1008'").show(false)/*** 这是所有dt=9999的*/val f_9999: DataFrame = spark.sql("""||select|nvl(id2,id) as id|,nvl(org_name2,org_name) as org_name|,nvl(parent_id2,parent_id) as parent_id|,nvl(sort2,sort) as sort|,nvl(org_type2,org_type) as org_type|,nvl(org_level2,org_level) as org_level|,nvl(is_auth_scope2,is_auth_scope) as is_auth_scope|,nvl(parent_auth_scope_id2,parent_auth_scope_id) as parent_auth_scope_id|,nvl(status2,status) as status|,nvl(icon_class2,icon_class) as icon_class|,nvl(create_id2,create_id) as create_id|,nvl(create_time2,create_time) as create_time|,nvl(update_id2,update_id) as update_id|,nvl(update_time2,update_time) as update_time|,nvl(version2,version) as version|,nvl(t_start2,t_start) as t_start|,nvl(t_end2,t_end) as t_end|,nvl(t_end2,t_end) as dt||from v1|||""".stripMargin)/*** +----+--------------------------------------------+---------+----+--------+---------+-------------+--------------------+------+-------------------------+---------+-------------------+---------+-------------------+-------+----------+----------+----------+* |id  |org_name                                    |parent_id|sort|org_type|org_level|is_auth_scope|parent_auth_scope_id|status|icon_class               |create_id|create_time        |update_id|update_time        |version|t_start   |t_end     |dt        |* +----+--------------------------------------------+---------+----+--------+---------+-------------+--------------------+------+-------------------------+---------+-------------------+---------+-------------------+-------+----------+----------+----------+* |1   |运营系统                                    |0        |0   |4       |1        |N            |null                |1     |iconfont icon-xitong     |655      |2019-05-20 17:58:11|null     |null               |null   |2023-01-31|9999-12-31|9999-12-31|*/println("f_9999 show")f_9999.show(false)println(s"在${dt}的发生状态变化的,新的有效区间[$dt,$lastDay]...")f_9999.filter(s"t_start='$dt'").show()f_9999.groupBy("dt").agg(count("id")).show()/*** 过期的数据* 需要闭合t_end* dt天发现有变化，那么则在dt-1天过期* 过期的数据：上一次的起始时间，必然小于dt(这个条件很重要，否则幂等计算会有问题，会把计算过的历史分区的起始时间给覆盖掉)*/val f_expire = spark.sql(s"""||select|id,|org_name,|parent_id,|sort,|org_type,|org_level,|is_auth_scope,|parent_auth_scope_id,|status,|icon_class,|create_id,|create_time,|update_id,|update_time,|version,|t_start,|cast(date_add('${dt}',-1) as string) as t_end,|cast(date_add('${dt}',-1) as string) as dt||from v1|where id2 is not null and id is not null and t_start<'$dt'||""".stripMargin)println("f_expire show")f_expire.show(false)/*** 没有动态分区，那就分别各自持久化*/if (!f_9999.isEmpty) {SinkUtil.sink_to_hive(lastDay, spark, f_9999, hive_db, hive_table = "ods_uac_org_zip", "parquet", MySaveMode.OverWriteByDt)}if (!f_expire.isEmpty) {SinkUtil.sink_to_hive(theDayBeforeDt, spark, f_expire, hive_db, hive_table = "ods_uac_org_zip", "parquet", MySaveMode.OverWriteByDt)}}
}

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

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

相关文章

C#调用Windows系统自带虚拟键盘的方法

C#调用Windows系统自带虚拟键盘的方法

C#调用Windows系统自带虚拟键盘&#xff0c;可以通过以下方法。 1、添加外部引用 private const Int32 WM_SYSCOMMAND 274;private const UInt32 SC_CLOSE 61536;[DllImport("user32.dll", SetLastError true, CharSet CharSet.Auto)]private static extern bo…
阅读更多...
目标检测YOLO实战应用案例100讲-基于机器视觉的水稻病虫害监测预警（续）

目标检测YOLO实战应用案例100讲-基于机器视觉的水稻病虫害监测预警（续）

目录 3.3 试验结果与分析 3.3.1 数据集介绍 3.3.3 评价标准 3.3.4 模型训练参数设置
阅读更多...
[Genode] ARM TrustZone

[Genode] ARM TrustZone

这是关于读文章ARM TrustZone的记录&#xff0c;原文是英文&#xff0c;刚开始会有点反应不过来&#xff0c;这里大部分是对文章的翻译与提取。 ARM信任区技术 ARM信任区是在 热烈讨论关于X86平台上的可信平台模块&#xff08;TPM&#xff09; 时引入的。。 就像TPM芯片神奇…
阅读更多...
【机器学习算法】机器学习：支持向量机（SVM）

【机器学习算法】机器学习：支持向量机（SVM）

转载自&#xff1a; 【精选】机器学习&#xff1a;支持向量机&#xff08;SVM&#xff09;-CSDN博客 1.概述 1.1&#xff0c;概念 支持向量机&#xff08;SVM&#xff09;是一类按监督学习方式对数据进行二元分类的广义线性分类器&#xff0c;其决策边界是对学习样本求解的最…
阅读更多...
这些来自各领域的全新机器人技术，你了解吗？

这些来自各领域的全新机器人技术，你了解吗？

原创 | 文 BFT机器人 01 人机交互的新工具 在人机交互领域&#xff0c;来自欧洲各地的研究人员开发了一种名为HEUROBOX的新工具&#xff0c;用于评估交互。HEUROBOX提供了84个基本启发式和228个高级启发式&#xff0c;用于评估人机交互的各个方面&#xff0c;如安全性、人体工…
阅读更多...
mac控制台命令小技巧

mac控制台命令小技巧

shigen日更文章的博客写手&#xff0c;擅长Java、python、vue、shell等编程语言和各种应用程序、脚本的开发。记录成长&#xff0c;分享认知&#xff0c;留住感动。 hello伙伴们&#xff0c;作为忠实的mac骨灰级别的粉丝&#xff0c;它真的给我带来了很多效率上的提升。那作为接…
阅读更多...
金蝶云星空和旺店通·旗舰奇门接口打通对接实战

金蝶云星空和旺店通·旗舰奇门接口打通对接实战

金蝶云星空和旺店通旗舰奇门接口打通对接实战 数据源系统:金蝶云星空 金蝶K/3Cloud&#xff08;金蝶云星空&#xff09;是移动互联网时代的新型ERP&#xff0c;是基于WEB2.0与云技术的新时代企业管理服务平台。金蝶K/3Cloud围绕着“生态、人人、体验”&#xff0c;旨在帮助企业…
阅读更多...
三极管与mos管的区分与应用

三极管与mos管的区分与应用

三极管与mos管的区分与应用 这部分知识每次用都要查,隔一段时间就忘记了.忍无可忍,再此写下来,作为之后然后查找的笔记 这里先推荐几篇我查阅到的笔记, <<三极管和MOS管驱动电路的正确用法>> https://blog.csdn.net/qq_21794157/article/details/122736035 <<…
阅读更多...
Azure 机器学习 - 搜索中的检索增强 (RAG)

Azure 机器学习 - 搜索中的检索增强 (RAG)

目录 一、Azure AI 信息检索系统介绍二、采用 Azure AI 搜索的 RAG 方法三、适合 Azure AI 搜索的自定义 RAG 模式四、Azure AI 搜索中的可搜索内容五、Azure AI 搜索中的内容检索构建查询响应按相关性排名适用于 RAG 方案的 Azure AI 搜索查询的示例代码 六、集成代码和 LLM七…
阅读更多...
无菌药品生产（A级区域）--气流流型测试可视化烟雾试验详细介绍

无菌药品生产（A级区域）--气流流型测试可视化烟雾试验详细介绍

技术背景 无菌药品是指法定药品标准中列有无菌检测项目的制剂和原料药&#xff0c;包括注射剂、眼用制剂、无菌软膏剂、无菌混悬剂等。目前工程中&#xff0c;以注射剂产品为主的厂房占据了很大的比重。 无菌药品生产质量风险管理中&#xff0c;人员及其活动被视为重大的污染…
阅读更多...
代码随想录算法训练营第一天|数组理论基础，704. 二分查找，27. 移除元素

代码随想录算法训练营第一天|数组理论基础，704. 二分查找，27. 移除元素

704. 二分查找 简单 给定一个 n 个元素有序的&#xff08;升序&#xff09;整型数组 nums 和一个目标值 target &#xff0c;写一个函数搜索 nums 中的 target&#xff0c;如果目标值存在返回下标&#xff0c;否则返回 -1。 示例 1: 输入: nums [-1,0,3,5,9,12], target 9 …
阅读更多...
云计算行业敲门砖—证书盘点

云计算行业敲门砖—证书盘点

未来10年&#xff0c;都会是云计算技术不断发展变革的时代&#xff0c;这其中会产生非常多的就业机会。有数据统计&#xff0c;未来五年&#xff0c;云计算行业人才缺口达150万&#xff0c;选对了行业&#xff0c;你就成功了一半。 云计算可以考的证书还是很多的&#xff0c;很…
阅读更多...
GZ033 大数据应用开发赛题第05套

GZ033 大数据应用开发赛题第05套

2023年全国职业院校技能大赛 赛题第05套 赛项名称&#xff1a; 大数据应用开发 英文名称&#xff1a; Big Data Application Development 赛项组别&#xff1a; 高等职业教育组 赛项编号&#xff1a; GZ033 …
阅读更多...
【Unity小技巧】图片使用的一些常见问题

【Unity小技巧】图片使用的一些常见问题

文章目录 前言Button不规则按钮点击空白区域不响应点击事件1. 设置资源参数2. 代码设置按钮Image的alphaHitTestMinimumThreshold3. 解释&#xff1a;4. 效果 Unity Image 原图比例控制方法一 Preserve Aspect1. 设置勾选Preserve Aspect&#xff08;保持长宽比&#xff09;&am…
阅读更多...
Python的requests库：解决文档缺失问题的策略与实践

Python的requests库：解决文档缺失问题的策略与实践

在Python的requests库中&#xff0c;有一个名为ALL_PROXY的参数&#xff0c;但是该参数的文档并未进行详细的描述。这使得用户在使用该参数时可能会遇到一些问题&#xff0c;例如不知道如何正确地配置和使用该参数。 解决方案 针对这个问题&#xff0c;我们可以采取以下几种解…
阅读更多...
小小发票拦住出海“巨头”，合合信息智能文档处理技术助力企业重塑财务管理流程

小小发票拦住出海“巨头”，合合信息智能文档处理技术助力企业重塑财务管理流程

作为连接企业采购、生产、运营、销售等经营行为的枢纽&#xff0c;财务系统的数字化、智能化升级近年来变得愈发重要。发票的录入、存储关乎企业应收账款管理和税务规范&#xff0c;通过技术手段提升发票管理的精准度与效率&#xff0c;也成为了财务数字化的关键环节。 近日&a…
阅读更多...
MySQL 8.0：性能优化与新功能

MySQL 8.0：性能优化与新功能

引言 MySQL 8.0是MySQL数据库管理系统的最新主要版本&#xff0c;带来了许多性能优化和新功能。本博客将探讨MySQL 8.0的性能提升以及一些使用技巧&#xff0c;助力你更好地利用这一强大的数据库系统。 新特性概览 MySQL 8.0相对于先前版本&#xff08;如MySQL 5.7及更早版本&a…
阅读更多...
如何解决 Critical dependency: the request of a dependency is an expression ？

如何解决 Critical dependency: the request of a dependency is an expression ？

更多ruoyi-nbcio功能请看演示系统 gitee源代码地址 前后端代码&#xff1a; ruoyi-nbcio: nbcio-boot的若依版本,基于ruoyi-flowable-plus和flowable6.7.2&#xff0c;目前处于开发功能完善阶段&#xff0c;目标是打造一个最好的若依平台上flowable流程管理系统开源版本&…
阅读更多...
CRM按行业细分的重要性

CRM按行业细分的重要性

很多企业和销售会诟病CRM系统不够贴合行业、功能也不够细分和实用。因为各行各业的业务千差万别&#xff0c;所以功能完备、使用满意度高的CRM一定是与不同行业业务场景高度匹配的&#xff0c;是深度行业化的。因此行业化是CRM发展的重要趋势之一&#xff0c;为什么CRM一定要走…
阅读更多...
【JS】鉴别项目是否处于嵌入模式

【JS】鉴别项目是否处于嵌入模式

文章目录 是否嵌入可以通过url的搜索参数去判断&#xff0c;例如新增参数embed <!-- 正常嵌入 --> <iframesrc"https://test.com/#/main?id66"width"100%"height"100%"frameBorder{0} /> <!-- 添加参数判断 --> <iframes…
阅读更多...
最新文章

Copyright @ 2022~2023