Flink深入浅出:JDBC Connector源码分析
大数据开发中,数据分析与报表制作是解析日常工作中最常遇到的任务。通常,源码我们通过读取Hive数据来进行计算,解析并将结果保存到数据库中,源码然后通过前端读取数据库来进行报表展示。解析dubboinvoker 源码然而,源码使用FlinkSQL可以简化这一过程,解析通过一个SQL语句即可完成整个ETL流程。源码
在Flink中,解析读取Hive数据并将数据写入数据库是源码常见的需求。本文将重点讲解数据如何写入数据库的解析过程,包括刷写数据库的源码机制和原理。
以下是解析本文将讲解的几个部分,以解答在使用过程中可能产生的源码疑问:
1. 表的定义
2. 定义的表如何找到具体的实现类(如何自定义第三方sink)
3. 写入数据的机制原理
(本篇基于1..0源码整理而成)
1. 表的定义
Flink官网提供了SQL中定义表的示例,以下以oracle为例:
定义好这样的表后,就可以使用insert into student执行插入操作了。接下来,我们将探讨其中的技术细节。
2. 如何找到实现类
实际上,这一过程涉及到之前分享过的SPI(服务提供者接口),即DriverManager去寻找Driver的过程。在Flink SQL执行时,会通过translate方法将SQL语句转换为对应的Operation,例如insert into xxx中的xxx会转换为CatalogSinkModifyOperation。这个操作会获取表的信息,从而得到Table对象。如果这个Table对象是CatalogTable,则会进入TableFactoryService.find()方法找到对应的实现类。
寻找实现类的过程就是SPI的过程。即通过查找路径下所有TableFactory.class的实现类,加载到内存中。这个SPI的定义位于resources下面的META-INFO下,定义接口以及实现类。
加载到内存后,首先判断是否是TableFactory的实现类,然后检查必要的参数是否满足(如果不满足会抛出异常,很多人在第一次使用Flink SQL注册表时,都会遇到NoMatchingTableFactoryException异常,其实都是因为配置的属性不全或者Jar报不满足找不到对应的TableFactory实现类造成的)。
找到对应的实现类后,调用对应的学校班级源码createTableSink方法就能创建具体的实现类了。
3. 工厂模式+创建者模式,创建TableSink
JDBCTableSourceSinkFactory是JDBC表的具体实现工厂,它实现了stream的sinkfactory。在1..0版本中,它不能在batch模式下使用,但在1.版本中据说会支持。这个类使用了经典的工厂模式,其中createStreamTableSink负责创建真正的Table,基于创建者模式构建JDBCUpsertTableSink。
创建出TableSink之后,就可以使用Flink API,基于DataStream创建一个Sink,并配置对应的并行度。
4. 消费数据写入数据库
在消费数据的过程中,底层基于PreparedStatement进行批量提交。需要注意的是提交的时机和机制。
控制刷写触发的最大数量 'connector.write.flush.max-rows' = ''
控制定时刷写的时间 'connector.write.flush.interval' = '2s'
这两个条件先到先触发,这两个参数都是可以通过with()属性配置的。
JDBCUpsertFunction很简单,主要的工作是包装对应的Format,执行它的open和invoke方法。其中open负责开启连接,invoke方法负责消费每条数据提交。
接下来,我们来看看关键的format.open()方法:
接下来就是消费数据,执行提交了
AppendWriter很简单,只是对PreparedStatement的封装而已
5. 总结
通过研究代码,我们应该了解了以下关键问题:
1. JDBC Sink执行的机制,比如依赖哪些包?(flink-jdbc.jar,这个包提供了JDBCTableSinkFactory的实现)
2. 如何找到对应的实现?基于SPI服务发现,扫描接口实现类,通过属性过滤,最终确定对应的实现类。
3. 底层如何提交记录?目前只支持append模式,底层基于PreparedStatement的addbatch+executeBatch批量提交
4. 数据写入数据库的时机和机制?一方面定时任务定时刷新,另一方面数量超过限制也会触发刷新。
更多Flink内容参考:
大数据笔试真题集锦---第五章:Hive面试题
我会不间断地更新维护,希望对正在寻找大数据工作的朋友们有所帮助。 第五章目录 第五章 Hive 5.1 Hive 运行原理(源码级) 1.1 reduce端join 在reduce端,对两个表的数据分别标记tag,发送数据。超级商城源码根据分区分组规则获取相同key的数据,再根据tag进行join操作,完成实际连接。 1.2 map端join 将小表复制到每个map task的内存中,仅扫描大表,对大表中key在小表中存在时进行join操作。使用DistributedCache.addCacheFile设置小表,通过标准IO获取数据。 1.3 semi join 先将参与join的表1的key复制到表3中,复制多份到各map task,过滤不在新表3的表2数据,最后进行reduce。 5.2 Hive 建表5.3.1 传统方式建表
定义数据类型,如:TINYINT, STRING, TIMESTAMP, DECIMAL。 使用ARRAY, MAP, STRUCT结构。5.3.2 CTAS查询建表
创建表时指定表名、存储格式、数据来源查询语句。 缺点:默认数据类型范围限制。5.3.3 Like建表
通过复制已有表的结构来创建新表。5.4 存储格式和压缩格式
选择ORC+bzip/gzip作为源存储,ORC+Snappy作为中间存储。 分区表单文件不大采用gzip压缩,桶表使用bzip或lzo支持分片压缩。 设置压缩参数,如"orc.compress"="gzip"。5.5 内部表和外部表
外部表使用external关键字和指定HDFS目录创建。 内部表在创建时生成对应目录的文件夹,外部表以指定文件夹为数据源。 内部表删除时删除整个文件夹,外部表仅删除元数据。5.6 分区表和分桶表
分区表按分区字段拆分存储,避免全表查询,提高效率。 动态分区通过设置参数开启,根据字段值决定分区。 分桶表依据分桶字段hash值分组拆分数据。5.7 行转列和列转行
行转列使用split、explode、laterview,列转行使用concat_ws、周易起名源码collect_list/set。5.8 Hive时间函数
from_unixtime、unix_timestamp、to_date、month、weekofyear、quarter、trunc、current_date、date_add、date_sub、datediff。 时间戳支持转换和截断,标准格式为'yyyy-MM-dd HH:mm:ss'。 month函数基于标准格式截断,识别时截取前7位。5.9 Hive 排名函数
row_number、dense_rank、rank。5. Hive 分析函数:Ntile
效果:排序并分桶。 ntile(3) over(partition by A order by B)效果,可用于取前%数据统计。5. Hive 拉链表更新
实现方式和优化策略。5. Hive 排序
order by、order by limit、sort by、sort by limit的原理和应用场景。5. Hive 调优
减少distinct、优化map任务数量、并行度优化、小文件问题解决、存储格式和压缩格式设置。5. Hive和Hbase区别
Hive和Hbase的区别,Hive面向分析、高延迟、结构化,Hbase面向编程、低延迟、非结构化。5. 其他
用过的开窗函数、表join转换原理、同城网源码sort by和order by的区别、交易表查询示例、登录用户数量查询、动态分区与静态分区的区别。mimikatz源码分析-lsadump模块(注册表)
mimikatz是一款内网渗透中的强大工具,本文将深入分析其lsadump模块中的sam部分,探索如何从注册表获取用户哈希。
首先,简要了解一下Windows注册表hive文件的结构。hive文件结构类似于PE文件,包括文件头和多个节区,每个节区又有节区头和巢室。其中,巢箱由HBASE_BLOCK表示,巢室由BIN和CELL表示,整体结构被称为“储巢”。通过分析hive文件的结构图,可以更直观地理解其内部组织。
在解析过程中,需要关注的关键部分包括块的签名(regf)和节区的签名(hbin)。这些签名对于定位和解析注册表中的数据至关重要。
接下来,深入解析mimikatz的解析流程。在具备sam文件和system文件的情况下,主要分为以下步骤:获取注册表system的句柄、读取计算机名和解密密钥、获取注册表sam的句柄以及读取用户名和用户哈希。若无sam文件和system文件,mimikatz将直接通过官方API读取本地机器的注册表。
在mimikatz中,会定义几个关键结构体,包括用于标识操作的注册表对象和内容的结构体(PKULL_M_REGISTRY_HANDLE)以及注册表文件句柄结构体(HKULL_M_REGISTRY_HANDLE)。这些结构体包含了文件映射句柄、映射到调用进程地址空间的位置、巢箱的起始位置以及用于查找子键和子键值的键巢室。
在获取注册表“句柄”后,接下来的任务是获取计算机名和解密密钥。密钥位于HKLM\SYSTEM\ControlSet\Current\Control\LSA,通过查找键值,将其转换为四个字节的密钥数据。利用这个密钥数据,mimikatz能够解析出最终的密钥。
对于sam文件和system文件的操作,主要涉及文件映射到内存的过程,通过Windows API(CreateFileMapping和MapViewOfFile)实现。这些API使得mimikatz能够在不占用大量系统资源的情况下,方便地处理大文件。
在获取了注册表系统和sam的句柄后,mimikatz会进一步解析注册表以获取计算机名和密钥。对于密钥的获取,mimikatz通过遍历注册表项,定位到特定的键值,并通过转换宽字符为字节序列,最终组装出密钥数据。
接着,解析过程继续进行,获取用户名和用户哈希。在解析sam键时,mimikatz首先会获取SID,然后遍历HKLM\SAM\Domains\Account\Users,解析获取用户名及其对应的哈希。解析流程涉及多个步骤,包括定位samKey、获取用户名和用户哈希,以及使用samKey解密哈希数据。
对于samKey的获取,mimikatz需要解密加密的数据,使用syskey作为解密密钥。解密过程根据加密算法(rc4或aes)有所不同,但在最终阶段,mimikatz会调用系统函数对数据进行解密,从而获取用户哈希。
在完成用户哈希的解析后,mimikatz还提供了一个额外的功能:获取SupplementalCreds。这个功能可以解析并解密获取对应用户的SupplementalCredentials属性,包括明文密码及哈希值,为用户提供更全面的哈希信息。
综上所述,mimikatz通过解析注册表,实现了从系统中获取用户哈希的高效功能,为内网渗透提供了强大的工具支持。通过深入理解其解析流程和关键结构体的定义,可以更好地掌握如何利用mimikatz进行深入的安全分析和取证工作。
beehive 源码阅读- go 语言的自动化机器
beehive源码深入解析:Go语言中的自动化机器设计
beehive的核心模块系统在包<p>bees</p>中体现其独特的解耦设计,这使得系统操作简便且易于扩展。只需要少量的学习,就能扩展自己的beehive功能。这里的"bee"代表Worker,执行具体任务,类似于采蜜的工蜂;而"hive"则是一个WorkerPool的工厂,通过简单配置(如一个token)即可创建针对特定任务的bee。
"chain"是连接事件和处理的关键,它将事件(如博客更新)与响应(如发送邮件)关联起来,通过事件通道(eventChan)触发并执行相应的action。WebBee的实现展示了如何在Run方法中接收事件并唤醒相应的bee,同时ServeHTTP函数负责http请求处理,暴露API供外部调用。
事件(Event)的处理通过<p>handleEvents</p>函数实现,它接收事件并将事件与对应的bee关联,进一步通过chains链接Event和Action,实现bee间的协作。Action的执行由<p>execAction</p>函数负责,可以处理预设选项或运行时传入的选项。
总的来说,beehive的自动化机器设计通过巧妙的解耦、事件驱动和灵活的链式处理,提供了一种高效且可扩展的编程模式。
Hive 编程专题之 - 自定义函数 Java 篇
Hive函数分为内置函数与自定义函数,内建函数包括字符、数值、日期与转换等类型。
自定义函数类似于传统商业数据库中的编译函数,如SQL Server中使用C#解决内建函数无法解决的问题,Oracle中则使用Java编写的Jar扩展功能,Hive中的自定义函数同样依赖Jar,提供Java编写程序以处理内置函数无法达到的功能。
使用Java编写Hive自定义函数步骤包括:
1. 常看所有内置与自定义函数。
2. Java或Scala编写自定义函数。
3. 使用Eclipse或其他Java编辑工具生成JAR文件。
4. 将生成的JAR文件放置于HDFS中,Hive即可应用。
5. 使用Java编写简单的Hive自定义函数,步骤如下:
5.1 使用Eclipse建立Maven项目。
5.2 引入特定的Hive/Hadoop JAR。
5.3 从Hive源代码或Hadoop基类库中寻找所需库。
5.4 编写简单的大写转换函数。
5.5 导出Eclipse,导入Hive类路径。
5.6 定义Hive函数,需带上全路径,即类的包路径。
5.7 修改Java代码,再执行相关步骤。
通过以上步骤,成功使用Java编写一个供Hive调用的函数。
spark sql源码系列 | with as 语句真的会把查询的数据存内存嘛?
在探讨 Spark SQL 中 with...as 语句是否真的会把查询的数据存入内存之前,我们需要理清几个关键点。首先,网上诸多博客常常提及 with...as 语句会将数据存放于内存中,来提升性能。那么,实际情况究竟如何呢?
让我们以 hive-sql 的视角来解答这一问题。在 hive 中,有一个名为 `hive.optimize.cte.materialize.threshold` 的参数。默认情况下,其值为 -1,代表关闭。当值大于 0 时(如设置为 2),with...as 语句生成的表将在被引用次数达到设定值后物化,从而确保 with...as 语句仅执行一次,进而提高效率。
接下来,我们通过具体测试来验证上述结论。在不调整该参数的情况下,执行计划显示 test 表被读取了两次。此时,我们将参数调整为 `set hive.optimize.cte.materialize.threshold=1`,执行计划显示了 test 表被物化的情况,表明查询结果已被缓存。
转而观察 Spark SQL 端,我们并未发现相关优化参数。Spark 对 with...as 的操作相对较少,在源码层面,通过获取元数据时所做的参数判断(如阈值与 cte 引用次数),我们可以发现 Spark 在这个逻辑上并未提供明确的优化机制,来专门针对 with...as 语句进行高效管理。
综上所述,通过与 hive-sql 的对比以及深入源码分析,我们得出了 with...as 语句在 Spark SQL 中是否把数据存入内存的结论,答案并不是绝对的。关键在于是否通过参数调整来物化结果,以及 Spark 在自身框架层面并未提供特定优化策略来针对 with...as 语句进行内存管理。因此,正确使用 with...as 语句并结合具体业务场景,灵活调整优化参数策略,是实现性能提升的关键。
通过深挖Clickhouse源码,我精通了数据去重!
数据去重的Clickhouse探索
在大数据面试中,数据去重是一个常考问题。虽然很多博主已经分享过相关知识,但本文将带您深入理解Hive引擎和Clickhouse在去重上的差异,尤其是后者如何通过MergeTree和高效的数据结构优化去重性能。Hive去重
Hive中,distinct可能导致数据倾斜,而group by则通过分布式处理提高效率。面试时,理解MapReduce的数据分区分组是关键。然而,对于大规模数据,Hive的处理速度往往无法满足需求。Clickhouse的登场
面对这个问题,Clickhouse凭借其列存储和MergeTree引擎崭露头角。MergeTree的高效体现在它的数据分区和稀疏索引,以及动态生成和合并分区的能力。Clickhouse:Yandex开源的实时分析数据库,每秒处理亿级数据
MergeTree存储结构:基于列存储,通过合并树实现高效去重
数据分区和稀疏索引
Clickhouse的分区策略和数据组织使得去重更为快速。稀疏索引通过标记大量数据区间,极大地减少了查询范围,提高性能。优化后的去重速度
测试显示,Clickhouse在去重任务上表现出惊人速度,特别是通过Bitmap机制,去重性能进一步提升。源码解析与原则
深入了解Clickhouse的底层原理,如Bitmap机制,对于优化去重至关重要,这体现了对业务实现性能影响的深度理解。总结与启示
对于数据去重,无论面试还是日常工作中,深入探究和实践是提升的关键。不断积累和学习,即使是初入职场者也能在大数据领域找到自己的位置。从线上某应用多作业并发创建同一HIVE表分区偶现失败问题聊起-深度剖析下HIVE创建表分区的内部逻辑
在大数据领域,早期版本的原生HIVE由于缺乏锁和事务机制,不支持并发写操作。多作业并发创建同一表分区或写数据到同一表分区时,易遇报错或数据不一致问题。HIVE1.X版本开始加强事务和锁支持,大幅降低并发写问题。实践中,避免同一表分区的并发写以确保应用一致性。本文分享线上多作业并发写同一表分区时报错问题,提供解决方法。
公司内部使用数据同步工具datago,用于增强datax并自动清理表分区、创建表分区。在使用星环TDH平台时,某客户生产环境在多分片并发采集时,datago作业运行失败。问题偶发,需查明底层原因及优化措施。具体报错为HDFS目录不存在。
问题出现在datago采集作业初始化阶段,涉及自动清理旧文件和自动创建表分区配置。异常堆栈与datago源码分析揭示问题原因。解决策略需避免并发创建同一表分区,并避免创建时list分区目录文件。由于无法调整HIVE表分区创建逻辑,需在自动创建表分区上进行调整。
本文深入剖析HIVE创建表分区内部逻辑,提供解决问题的思路。对HIVE源码感兴趣者可进一步探索相关方法。
2024-12-22 22:35
2024-12-22 21:24
2024-12-22 20:53
2024-12-22 20:36
2024-12-22 20:18