Hudi Clean 清理文件实现分析

2022-04-18

前言

源码层面总结分析Hudi Clean是如何实现的，不了解Hudi Clean的可以先看这篇：一文彻底理解Apache Hudi的清理服务。
Hudi Clean主要是清理删除不需要的历史文件，可以根据实际业务需要配置参数，不能影响查询，比如某个查询语句正在用某个文件，Clean如果删除了这个文件，查询就会报错。
这里只是删除历史文件，Hudi的文件是有多个版本的，不管配置什么参数，使用什么策略，都不会删除当前最新版本的文件。
Hudi 0.9.0版本有两种清理策略KEEP_LATEST_COMMITS和KEEP_LATEST_FILE_VERSIONS,默认为KEEP_LATEST_COMMITS
KEEP_LATEST_COMMITS：简单讲就是根据commit的次数，默认保留最新的10个commit的所有文件，对于10个之前的文件只保留最新版本的文件，历史文件全部删除
KEEP_LATEST_FILE_VERSIONS：简单讲就是保留文件的版本数，默认保留三个版本
具体的可以看上面的那篇公众号文章

目前最新版本0.11.0 添加了一个新的策略KEEP_LATEST_BY_HOURS:根据小时数清理，默认保留最近24小时的文件，具体实现请查看PR：[HUDI-349] Added new cleaning policy based on number of hours

本文以Hudi 0.9.0 Java Client COW表进行分析

more >>

展开全文 >>

Kyuubi 安装配置总结

2022-03-25

前言

前段时间研究了一下Kyuubi，主要是安装配置，适配Spark版本，验证Spark Server HA 的功能，基本都验证通过，但是后续没有实际使用，现在回忆总结一下，避免遗忘。
主要适配Spark2.4.5 以及 Spark3.1.2版本，同时验证是否支持Hudi。

版本说明

目前Kyuubi最新版本为1.4，Kyuubi 1.x 默认不支持Spark2，1.4版本默认Spark版本3.1.2,并且默认支持Hudi,但是因hudi0.9版本不支持Spark3.1.2，所以需要hudi0.10.1
hudi0.10.1 已经发布，可以通过 Mvn 下载：https://repo1.maven.org/maven2/org/apache/hudi/hudi-spark3.1.2-bundle_2.12/0.10.1/hudi-spark3.1.2-bundle_2.12-0.10.1.jar
也可以自己打包
要想支持Spark2可以选择Kyuubi0.7，两个Kyuubi 版本都支持HA，但是0.7版本默认不支持hudi

0.7版本打包: git 切换到branch-0.7
修改pom,添加

<profile>
    <id>spark-2.4.5</id>
    <properties>
        <spark.version>2.4.5</spark.version>
        <scalatest.version>3.0.3</scalatest.version>
    </properties>
</profile>

然后执行打包命令

1	./build/dist --tgz -P spark-2.4.5

打包完成后，生成kyuubi-0.7.0-SNAPSHOT-bin-spark-2.4.5.tar.gz

more >>

展开全文 >>

SparkSQL JDBC 查询Oracle、MySQL 转化为Hudi表

2022-02-15

前言

本文总结如何只用SQL迁移关系型数据库中的表转化为Hudi表，这样的意义在于方便项目上对Spark、Hudi等不熟悉的人员运维。

Spark Thrift Server

首先将Orace的驱动拷贝至Spark jars目录下

启动Spark Thrift Server 扩展支持Hudi

/usr/hdp/3.1.0.0-78/spark2/bin/spark-submit --master yarn --deploy-mode client --executor-memory 2G --num-executors 6 --executor-cores 2 --driver-memory 4G --driver-cores 2 --class org.apache.spark.sql.hive.thriftserver.HiveThriftServer2 --name Thrift-20003 --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'  --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' --principal spark/indata-192-168-44-128.indata.com@INDATA.COM --keytab /etc/security/keytabs/spark.service.keytab --hiveconf hive.server2.thrift.http.port=20003 >~/server.log 2>&1 &

Beeline连接

/usr/hdp/3.1.0.0-78/spark2/bin/beeline -u "jdbc:hive2://192.168.44.128:20003/hudi;principal=HTTP/indata-192-168-44-128.indata.com@INDATA.COM?hive.server2.transport.mode=http;hive.server2.thrift.http.path=cliservice"

more >>

展开全文 >>

Spark Thrift Server HA 解决方案

2022-02-11

前言

本文为组内同事整理,这里稍作改动记录。本文主要修改Spark源码，实现Spark Spark Thrift Server 注册到到ZK，通过ZK连接实现负载均衡；另外可以通过使用Kyuubi实现HA，这里不做详细描述

背景

Spark ThriftServer 不支持Zookeeper连接，不能实现负载均衡。

解决方案:修改Spark ThriftServer源码，使其支持Zookeeper连接

版本信息：Spark3.1.2 ；Hive2.3.7

Spark ThriftServer 启动命令

nohup spark-submit --master yarn --deploy-mode client --num-executors 2 --executor-cores 1 --executor-memory 1G --class org.apache.spark.sql.hive.thriftserver.HiveThriftServer2 --name SparkThriftServer_1 spark-internal --hiveconf hive.server2.thrift.http.port=20003 >>  /var/log/spark2/20003.log 2>&1 < /dev/null &

Spark 源码debug

./bin/spark-submit --master yarn --deploy-mode client --num-executors 2 --executor-cores 1 --executor-memory 1G --class org.apache.spark.sql.hive.thriftserver.HiveThriftServer2 --name SparkThriftServer_1 spark-internal --hiveconf hive.server2.thrift.http.port=20003  --driver-java-options "-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005"

Spark 连接

单节点

1	!connect jdbc:hive2://indata-1192-168-44-128:20003/default;principal=HTTP/indata-1192-168-44-128@INDATA.COM?hive.server2.transport.mode=http;hive.server2.thrift.http.path=cliservice;

Zookeeper

1	!connect jdbc:hive2://192-168-44-129.indata.com:2181,indata-1192-168-44-128:2181,indata-192-168-44-130.indata.com:2181/default;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2_zk

more >>

展开全文 >>

Presto查询Hudi异常解决

2021-12-23

前言

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

异常

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 >>

展开全文 >>