Presto查询Hudi异常解决

前言

记录一个Presto查询Hudi的异常的解决办法，本人目前对Presto还不是很熟，只是在使用过程中遇到了问题，记录一下异常解决方法以及过程

异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
2021-12-22T17:29:55.440+0800    ERROR   SplitRunner-101-126     com.facebook.presto.execution.executor.TaskExecutor     Error processing Split 20211222_092954_00047_8xk77.1.0.0-0 {path=hdfs://cluster1/warehouse/tablespace/managed/hive/test_dkl.db/test_hudi_1/069eddc2-f3bf-4efc-911b-3cb3aa523a8e-0_0-0-0_20211222172800.parquet, start=0, length=435116, fileSize=435116, fileModifiedTime=1640165281695, hosts=[], database=test_dkl, table=test_hudi_1, nodeSelectionStrategy=NO_PREFERENCE, partitionName=<UNPARTITIONED>, s3SelectPushdownEnabled=false, cacheQuotaRequirement=CacheQuotaRequirement{cacheQuotaScope=GLOBAL, quota=Optional.empty}} (start = 9.271133085962032E9, wall = 49 ms, cpu = 0 ms, wait = 0 ms, calls = 1)
org.apache.parquet.io.ParquetDecodingException: Can not read value at 0 in block -1 in file hdfs://cluster1/warehouse/tablespace/managed/hive/test_dkl.db/test_hudi_1/069eddc2-f3bf-4efc-911b-3cb3aa523a8e-0_0-0-0_20211222172800.parquet
        at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:251)
        at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:207)
        at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:98)
        at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.<init>(ParquetRecordReaderWrapper.java:60)
        at org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat.getRecordReader(MapredParquetInputFormat.java:92)
        at org.apache.hudi.hadoop.HoodieParquetInputFormat.getRecordReader(HoodieParquetInputFormat.java:216)
        at com.facebook.presto.hive.HiveUtil.createRecordReader(HiveUtil.java:266)
        at com.facebook.presto.hive.GenericHiveRecordCursorProvider.lambda$createRecordCursor$0(GenericHiveRecordCursorProvider.java:74)
        at com.facebook.presto.hive.authentication.UserGroupInformationUtils.lambda$executeActionInDoAs$0(UserGroupInformationUtils.java:29)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:360)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1726)
        at com.facebook.presto.hive.authentication.UserGroupInformationUtils.executeActionInDoAs(UserGroupInformationUtils.java:27)
        at com.facebook.presto.hive.authentication.DirectHdfsAuthentication.doAs(DirectHdfsAuthentication.java:38)
        at com.facebook.presto.hive.HdfsEnvironment.doAs(HdfsEnvironment.java:81)
        at com.facebook.presto.hive.GenericHiveRecordCursorProvider.createRecordCursor(GenericHiveRecordCursorProvider.java:73)
        at com.facebook.presto.hive.HivePageSourceProvider.createHivePageSource(HivePageSourceProvider.java:478)
        at com.facebook.presto.hive.HivePageSourceProvider.createPageSource(HivePageSourceProvider.java:184)
        at com.facebook.presto.spi.connector.classloader.ClassLoaderSafeConnectorPageSourceProvider.createPageSource(ClassLoaderSafeConnectorPageSourceProvider.java:63)
        at com.facebook.presto.split.PageSourceManager.createPageSource(PageSourceManager.java:80)
        at com.facebook.presto.operator.TableScanOperator.getOutput(TableScanOperator.java:249)
        at com.facebook.presto.operator.Driver.processInternal(Driver.java:424)
        at com.facebook.presto.operator.Driver.lambda$processFor$9(Driver.java:307)
        at com.facebook.presto.operator.Driver.tryWithLock(Driver.java:728)
        at com.facebook.presto.operator.Driver.processFor(Driver.java:300)
        at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:1077)
        at com.facebook.presto.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:162)
        at com.facebook.presto.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:599)
        at com.facebook.presto.$gen.Presto_0_265_1_ad1fce6____20211222_010139_1.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.UnsupportedOperationException: readDirect unsupported in RemoteBlockReader
        at org.apache.hadoop.hdfs.RemoteBlockReader.read(RemoteBlockReader.java:492)
        at org.apache.hadoop.hdfs.DFSInputStream$ByteBufferStrategy.doRead(DFSInputStream.java:789)
        at org.apache.hadoop.hdfs.DFSInputStream.readBuffer(DFSInputStream.java:823)
        at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSInputStream.java:883)
        at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:938)
        at org.apache.hadoop.fs.FSDataInputStream.read(FSDataInputStream.java:143)
        at org.apache.parquet.hadoop.util.H2SeekableInputStream$H2Reader.read(H2SeekableInputStream.java:81)
        at org.apache.parquet.hadoop.util.H2SeekableInputStream.readFully(H2SeekableInputStream.java:90)
        at org.apache.parquet.hadoop.util.H2SeekableInputStream.readFully(H2SeekableInputStream.java:75)
        at org.apache.parquet.hadoop.ParquetFileReader$ConsecutiveChunkList.readAll(ParquetFileReader.java:1174)
        at org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:805)
        at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:127)
        at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:222)
more >>

展开全文 >>

利用Submarin集成Spark-Ranger

前言

本文总结如何利用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 总结(二)-源码分析

前言

在上一篇博客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 总结

前言

总结 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默认使用布隆索引，布隆索引只保证同一分区下同一个主键对应的值唯一,可以使用全局索引保证所有分区值唯一，这里不展开细说

1
2
3
4
5
6
7
8
9
10
11
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。

1
2
3
4
5
6
7
8
9
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 添加列总结

前言

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

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

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~