利用Submarin集成Spark-Ranger

2021-12-02

前言

本文总结如何利用Submarin集成Spark-Ranger，通过ranger控制Spark SQL的权限

前提

已经安装了Spark、Hive、kerberos、Ranger,并且Hive已经集成了Ranger,本文环境基于Ambari

submarine-spark-security 插件打包

官网文档https://submarine.apache.org/docs/userDocs/submarine-security/spark-security/README (0.6.0版本)

1	git clone https://github.com/apache/submarine

当前master分支（0.7.0）已经没有submarine-spark-security模块了，需要切换到tag:release-0.6.0-RC0

然后利用mvn命令打包

1	mvn clean package -Dmaven.javadoc.skip=true -DskipTests -pl :submarine-spark-security -Pspark-2.4 -Pranger-1.2

打的submarine-spark-security-0.6.0.jar在目录submarine\submarine-security\spark-security\target

然后将submarine-spark-security-0.6.0.jar上传到$SPARK_HOME/jars

more >>

展开全文 >>

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
}