Hudi preCombinedField 总结(二)-源码分析

2021-11-30

前言

在上一篇博客Hudi preCombinedField 总结中已经对preCombinedField进行总结过一次了，由于当时对源码理解还不够深入，导致分析的不全面，现在对源码有了进一步的理解，所以再进行总结补充一下。

历史比较值

上面总结中：

DF：无论新记录的ts值是否大于历史记录的ts值，都会覆盖写，直接更新。
SQL:写数据时，ts值大于等于历史ts值，才会更新，小于历史值则不更新。

这里解释一下原因，首先Spark SQL PAYLOAD_CLASS_NAME 默认值为ExpressionPayload，而ExpressionPayload继承了DefaultHoodieRecordPayload

1
2
3

class ExpressionPayload(record: GenericRecord,
                        orderingVal: Comparable[_])
  extends DefaultHoodieRecordPayload(record, orderingVal) {

DefaultHoodieRecordPayload 里的needUpdatingPersistedRecord实现了历史值进行比较，具体实现，后面会进行分析

而 Spark DF在hudi0.9.0版本 PAYLOAD_CLASS_NAME的默认值为OverwriteWithLatestAvroPayload，它是DefaultHoodieRecordPayload的父类并没有实现和历史值进行比较

more >>

展开全文 >>

Hudi preCombinedField 总结

2021-07-10

前言

总结 HUDI preCombinedField,分两大类总结，一类是Spark SQL,这里指的是merge，因为只有merge语句中有多条记录，讨论preCombinedField才有意义；一类是Spark DF，HUDI0.9版本支持SQL建表和增删改查

总结

先说结论：

Spark DF建表写数据时（含更新）：
1、UPSERT，当数据重复时（这里指同一主键对应多条记录），程序在写数据前会根据预合并字段ts进行去重，去重保留ts值最大的那条记录，且无论新记录的ts值是否大于历史记录的ts值，都会覆盖写，直接更新。
2、INSERT时，没有预合并，程序依次写入，实际更新为最后一条记录，且无论新记录的ts值是否大于历史记录的ts值，都会覆盖写，直接更新。
Spark SQL建表，写数据时（含更新）：
有ts时，预合并时如果数据重复取预合并字段值最大的那条记录，最大值相同的取第一个。写数据时，ts值大于等于历史ts值，才会更新，小于历史值则不更新。
没有ts时，则默认将主键字段的第一个值作为预合并字段，如果数据重复，去重时会取第一个值，写数据时，直接覆盖历史数据（因为这里的预合并字段为主键字段，等于历史值，其实原理跟上面有ts时一样）
1
PRECOMBINE_FIELD.key -> targetKey2SourceExpression.keySet.head, // set a default preCombine field

说明：
1、这里有ts代表设置了preCombinedField字段
2、hudi默认使用布隆索引，布隆索引只保证同一分区下同一个主键对应的值唯一,可以使用全局索引保证所有分区值唯一，这里不展开细说

private String getDefaultIndexType(EngineType engineType) {
  switch (engineType) {
    case SPARK:
      return HoodieIndex.IndexType.BLOOM.name();
    case FLINK:
    case JAVA:
      return HoodieIndex.IndexType.INMEMORY.name();
    default:
      throw new HoodieNotSupportedException("Unsupported engine " + engineType);
  }
}

3、如果在测试过程中，发现和我的结论不一致，可能和后面的注意事项有关。
4、当指定了hoodie.datasource.write.insert.drop.duplicates=true时，不管是insert还是upsert,如果存在历史数据则不更新。实际在源码中，如果为upsert，也会修改为insert。

if (hoodieConfig.getBoolean(INSERT_DROP_DUPS) &&
  operation == WriteOperationType.UPSERT) {

  log.warn(s"$UPSERT_OPERATION_OPT_VAL is not applicable " +
    s"when $INSERT_DROP_DUPS is set to be true, " +
    s"overriding the $OPERATION to be $INSERT_OPERATION_OPT_VAL")

  operation = WriteOperationType.INSERT
}

more >>

展开全文 >>

Spark DataFrame 添加列总结

2021-06-16

前言

因添加列在平时可能会经常用到，但是长时间不用，可能会忘记应该用哪个函数，这样再重新查找比较耽误时间，于是总结代码进行备忘。主要总结：

根据现有的列添加
添加自增ID
添加一列常量
添加当前时间
转换为timestamp类型
转换为date类型

more >>

展开全文 >>

通过DBeaver本地访问远程Kerberos环境下的Hive

2021-06-03

前言

本文讲解如何通过数据库客户端界面工具DBeaver连接远程Kerberos环境下的Hive。

因为在远程服务器上的命令行里写SQL查询Hive表，如果数据量和表字段比较多，命令行界面不利于分析表数据，所以需要客户端工具如DBeave远程连接Hive查询数据，但是DBeaver默认的不能访问Kerberos下的Hive，需要一些配置才可以访问，这里记录一下。

1、DBeaver连接Hive

关于DBeaver如何连接正常的不带Kerberos认证的Hive,请参考通过数据库客户端界面工具DBeaver连接Hive,本文只讲解如何在Windows上用DBeaver对kerberos认证

2、安装Kerberos客户端

官网下载https://web.mit.edu/kerberos/dist/index.html,我下载的是这个：Kerberos for Windows Release 4.1 - current release

下载下来后，一键安装

3、krb5.ini

我的kerberos客户端安装目录为D:\program\company\kerberos，将集群上的/etc/krb5.conf，复制到此目录下，改名为krb5.ini,内容如下：
more >>

展开全文 >>

Spark 本地连接远程服务器上带有kerberos认证的Hive

2021-05-19

前言

因为公司的测试环境带有kerberos，而我经常需要本地连接测试集群上的hive，以进行源码调试。而本地认证远程集群的kerberos，并访问hive，和在服务器上提交Spark程序代码有些不同，所以专门研究了一下
并进行总结。

服务器上

在服务器上提交Spark程序认证kerberos比较简单，有两种方法：

使用kinit 缓存票据 kinit -kt /etc/security/keytabs/hive.service.keytab hive/indata-192-168-44-128.indata.com@INDATA.COM，然后提交Spark程序即可
在spark-submit 中添加参数 –principal hive/indata-192-168-44-128.indata.com@INDATA.COM –keytab /etc/security/keytabs/hive.service.keytab

本地

本地连接，稍微复杂点，首先要配好环境，比如Hadoop的环境变量、winutils等，然后需要配置hosts,将服务器上的/etc/hosts里面的内容拷贝出来，粘贴Windows上的hosts文件里即可

代码

首先需要将集群上的hive-site.xml,core-site.xml,yarn-site.xml,hdfs-site.xml拷贝到src/main/resources文件夹中，其中hive-site.xml是为了连接hive,core-site.xml、hdfs-site.xml和yarn-site.xml是为了认证kerberos

more >>

展开全文 >>