华为云用户手册

Hive支持ZSTD压缩格式 ZSTD（全称为Zstandard）是一种开源的无损数据压缩算法，其压缩性能和压缩比均优于当前Hadoop支持的其他压缩格式，本特性使得Hive支持ZSTD压缩格式的表。Hive支持基于ZSTD压缩的存储格式有常见的ORC，RCFile，TextFile，JsonFile，Parquet，Squence， CS V。 本特性仅适用于 MRS 3.1.2及之后版本。 ZSTD压缩格式的建表方式如下： ORC存储格式建表时可指定TBLPROPERTIES("orc.compress"="zstd")： create table tab_1(...) stored as orc TBLPROPERTIES("orc.compress"="zstd"); Parquet存储格式建表可指定TBLPROPERTIES("parquet.compression"="zstd")： create table tab_2(...) stored as parquet TBLPROPERTIES("parquet.compression"="zstd"); 其他格式或通用格式建表可执行设置参数指定compress,codec为“org.apache.hadoop.io.compress.ZStandardCode”： set hive.exec.compress.output=true; set mapreduce.map.output.compress=true; set mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.ZStandardCodec; set mapreduce.output.fileoutputformat.compress=true; set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.ZStandardCodec; set hive.exec.compress.intermediate=true; create table tab_3(...) stored as textfile; ZSTD压缩格式的表和其他普通压缩表的SQL操作没有区别，可支持正常的增删查及聚合类SQL操作。 父主题： 使用Hive

Hive配置类问题 Hive SQL执行报错：java.lang.OutOfMemoryError: Java heap space. 解决方案： 对于MapReduce任务，增大下列参数： set mapreduce.map.memory.mb=8192; set mapreduce.map.java.opts=-Xmx6554M; set mapreduce.reduce.memory.mb=8192; set mapreduce.reduce.java.opts=-Xmx6554M; 对于Tez任务，增大下列参数： set hive.tez.container.size=8192; Hive SQL对列名as为新列名后，使用原列名编译报错：Invalid table alias or column reference 'xxx'. 解决方案：set hive.cbo.enable=true; Hive SQL子查询编译报错：Unsupported SubQuery Expression 'xxx': Only SubQuery expressions that are top level conjuncts are allowed. 解决方案：set hive.cbo.enable=true; Hive SQL子查询编译报错：CalciteSubquerySemanticException [Error 10249]: Unsupported SubQuery Expression Currently SubQuery expressions are only allowed as Where and Having Clause predicates. 解决方案：set hive.cbo.enable=true; Hive SQL编译报错：Error running query: java.lang.AssertionError: Cannot add expression of different type to set. 解决方案：set hive.cbo.enable=false; Hive SQL执行报错：java.lang.NullPointerException at org.apache.hadoop.hive.ql.udf.generic.GenericUDAFComputeStats$GenericUDAFNumericStatsEvaluator.init. 解决方案：set hive.map.aggr=false; Hive SQL设置hive.auto.convert.join = true（默认开启）和hive.optimize.skewjoin=true执行报错：ClassCastException org.apache.hadoop.hive.ql.plan.ConditionalWork cannot be cast to org.apache.hadoop.hive.ql.plan.MapredWork. 解决方案：set hive.optimize.skewjoin=false; Hive SQL设置hive.auto.convert.join=true（默认开启）、hive.optimize.skewjoin=true和hive.exec.parallel=true执行报错：java.io.FileNotFoundException: File does not exist:xxx/reduce.xml. 解决方案： 方法一：切换执行引擎为Tez，详情请参考切换Hive执行引擎为Tez。 方法二：set hive.exec.parallel=false; 方法三：set hive.auto.convert.join=false; Hive on Tez执行Bucket表Join报错：NullPointerException at org.apache.hadoop.hive.ql.exec.CommonMergeJoinOperator.mergeJoinComputeKeys 解决方案：set tez.am.container.reuse.enabled=false; 父主题： Hive常见问题

操作场景 HBase可以通过对HFile中的data block编码，减少keyvalue中key的重复部分，从而减少空间的使用。目前对data block的编码方式有：NONE、PREFIX、DIFF、FAST_DIFF和ROW_INDEX_V1，其中NONE表示不使用编码。另外，HBase还支持使用压缩算法对HFile文件进行压缩，默认支持的压缩算法有：NONE、GZ、SNAPPY和ZSTD，其中NONE表示HFile不压缩。 这两种方式都是作用在HBase的列簇上，可以同时使用，也可以单独使用。

Compaction Compaction用于合并mor表Base和Log文件。 对于Merge-On-Read表，数据使用列式Parquet文件和行式Avro文件存储，更新被记录到增量文件，然后进行同步/异步compaction生成新版本的列式文件。Merge-On-Read表可减少数据摄入延迟，因而进行不阻塞摄入的异步Compaction很有意义。 异步Compaction会进行如下两个步骤： 调度Compaction：由入湖作业完成，在这一步，Hudi扫描分区并选出待进行compaction的FileSlice，最后CompactionPlan会写入Hudi的Timeline。 执行Compaction：一个单独的进程/线程将读取CompactionPlan并对FileSlice执行Compaction操作。 使用Compaction的方式分为同步和异步两种： 同步方式由参数hoodie.compact.inline控制，默认为true，自动生成compaction调度计划并执行compaction： 关闭同步compaction datasource写入时可以通过 .option("hoodie.compact.inline", "false") 来关闭自动compaction。 spark-sql写入时可以通过set hoodie.compact.inline=false;来关闭自动compaction。 仅同步生成compaction调度而不执行compaction ·datasource写入时可以通过以下option参数来实现： option("hoodie.compact.inline", "true"). option("hoodie.schedule.compact.only.inline", "true"). option("hoodie.run.compact.only.inline", "false"). ·spark-sql写入时可以通过set 以下参数来实现： set hoodie.compact.inline=true; set hoodie.schedule.compact.only.inline=true; set hoodie.run.compact.only.inline=false; 异步方式由spark-sql来实现。 如果需要在异步compaction时只执行已经产生的compaction调度计划而不创建新的调度计划，则需要通过set命令设置以下参数： set hoodie.compact.inline=true; set hoodie.schedule.compact.only.inline=false; set hoodie.run.compact.only.inline=true; 更多compaction参数请参考compaction&cleaning配置章节。 为了保证入湖的最高效率，推荐使用同步产生compaction调度计划，异步执行compaction调度计划的方式。 父主题： 数据管理维护

HoodieDeltaStreamer流式写入 Hudi自带HoodieDeltaStreamer工具支持流式写入，也可以使用SparkStreaming以微批的方式写入。HoodieDeltaStreamer提供以下功能： 支持Kafka，DFS多种数据源接入 。 支持管理检查点、回滚和恢复，保证exactly once语义。 支持自定义转换操作。 示例： 准备配置文件kafka-source.properties #hudi配置hoodie.datasource.write.recordkey.field=idhoodie.datasource.write.partitionpath.field=agehoodie.upsert.shuffle.parallelism=100#hive confighoodie.datasource.hive_sync.table=hudimor_deltastreamer_partitionhoodie.datasource.hive_sync.partition_fields=agehoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractorhoodie.datasource.hive_sync.use_jdbc=falsehoodie.datasource.hive_sync.support_timestamp=true# Kafka Source topichoodie.deltastreamer.source.kafka.topic=hudimor_deltastreamer_partition#checkpointhoodie.deltastreamer.checkpoint.provider.path=hdfs://hacluster/tmp/huditest/hudimor_deltastreamer_partition# Kafka props# The kafka cluster we want to ingest frombootstrap.servers= xx.xx.xx.xx:xxauto.offset.reset=earliest#auto.offset.reset=latestgroup.id=hoodie-delta-streameroffset.rang.limit=10000 指定HoodieDeltaStreamer执行参数（具体参数配置，请查看官网https://hudi.apache.org/ ）执行如下命令： spark-submit --master yarn --jars /opt/hudi-java-examples-1.0.jar // 指定spark运行时需要的hudi jars路径 --driver-memory 1g --executor-memory 1g --executor-cores 1 --num-executors 2 --conf spark.kryoserializer.buffer.max=128m --driver-class-path /opt/client/Hudi/hudi/conf:/opt/client/Hudi/hudi/lib/*:/opt/client/Spark2x/spark/jars/*:/opt/hudi-examples-0.6.1-SNAPSHOT.jar:/opt/hudi-examples-0.6.1-SNAPSHOT-tests.jar // 指定spark driver需要的hudi jars路径 --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer spark-internal --props file:///opt/kafka-source.properties // 指定配置文件，注意：使用yarn-cluster模式提交任务时，请指定配置文件路径为HDFS路径。 --target-base-path /tmp/huditest/hudimor1_deltastreamer_partition // 指定hudi表路径 --table-type MERGE_ON_READ // 指定要写入的hudi表类型 --target-table hudimor_deltastreamer_partition // 指定hudi表名 --source-ordering-field name // 指定hudi表预合并列 --source-class org.apache.hudi.utilities.sources.JsonKafkaSource // 指定消费的数据源 为JsonKafkaSource， 该参数根据不同数据源指定不同的source类 --schemaprovider-class com.huawei.bigdata.hudi.examples.DataSchemaProviderExample // 指定hudi表所需要的schema --transformer-class com.huawei.bigdata.hudi.examples.TransformerExample // 指定如何处理数据源拉取来的数据，可根据自身业务需求做定制 --enable-hive-sync // 开启hive同步，同步hudi表到hive --continuous // 指定流处理模式为连续模式

compaction&cleaning配置 参数 描述 默认值 hoodie.clean.automatic 是否执行自动clean。 true hoodie.cleaner.policy 要使用的清理政策。Hudi将删除旧版本的parquet文件以回收空间。 任何引用此版本文件的查询和计算都将失败。需要确保数据保留的时间超过最大查询执行时间。 KEEP_LATEST_COMMITS hoodie.cleaner.commits.retained 保留的提交数。因此，数据将保留为num_of_commits * time_between_commits（计划的），这也直接转化为逐步提取此数据集的数量。 10 hoodie.keep.max.commits 触发归档操作的commit数阈值 30 hoodie.keep.min.commits 归档操作保留的commit数 20 hoodie.commits.archival.batch 这控制着批量读取并一起归档的提交即时的数量。 10 hoodie.parquet.small.file.limit 该值应小于maxFileSize，如果将其设置为0，会关闭此功能。由于批处理中分区中插入记录的数量众多，总会出现小文件。Hudi提供了一个选项，可以通过将对该分区中的插入作为对现有小文件的更新来解决小文件的问题。此处的大小是被视为“小文件大小”的最小文件大小。 104857600 byte hoodie.copyonwrite.insert.split.size 插入写入并行度。为单个分区的总共插入次数。写出100MB的文件，至少1KB大小的记录，意味着每个文件有100K记录。默认值是超额配置为500K。 为了改善插入延迟，请对其进行调整以匹配单个文件中的记录数。将此值设置为较小的值将导致文件变小（尤其是当compactionSmallFileSize为0时）。 500000 hoodie.copyonwrite.insert.auto.split Hudi是否应该基于最后24个提交的元数据动态计算insertSplitSize，默认关闭。 true hoodie.copyonwrite.record.size.estimate 平均记录大小。如果指定，Hudi将使用它，并且不会基于最后24个提交的元数据动态地计算。 没有默认值设置。这对于计算插入并行度以及将插入打包到小文件中至关重要。 1024 hoodie.compact.inline 当设置为true时，紧接在插入或插入更新或批量插入的提交或增量提交操作之后由摄取本身触发压缩。 true hoodie.compact.inline.max.delta.commits 触发内联压缩之前要保留的最大增量提交数。 5 hoodie.compaction.lazy.block.read 当CompactedLogScanner合并所有日志文件时，此配置有助于选择是否应延迟读取日志块。选择true以使用I/O密集型延迟块读取（低内存使用），或者为false来使用内存密集型立即块读取（高内存使用）。 true hoodie.compaction.reverse.log.read HoodieLogFormatReader会从pos=0到pos=file_length向前读取日志文件。如果此配置设置为true，则Reader会从pos=file_length到pos=0反向读取日志文件。 false hoodie.cleaner.parallelism 如果清理变慢，请增加此值。 200 hoodie.compaction.strategy 用来决定在每次压缩运行期间选择要压缩的文件组的压缩策略。默认情况下，Hudi选择具有累积最多未合并数据的日志文件。 org.apache.hudi.table.action.compact.strategy. LogFileSizeBasedCompactionStrategy hoodie.compaction.target.io LogFileSizeBasedCompactionStrategy的压缩运行期间要花费的MB量。当压缩以内联模式运行时，此值有助于限制摄取延迟。 500 * 1024 MB hoodie.compaction.daybased.target.partitions 由org.apache.hudi.io.compact.strategy.DayBasedCompactionStrategy使用，表示在压缩运行期间要压缩的最新分区数。 10 hoodie.compaction.payload.class 这需要与插入/插入更新过程中使用的类相同。就像写入一样，压缩也使用记录有效负载类将日志中的记录彼此合并，再次与基本文件合并，并生成压缩后要写入的最终记录。 org.apache.hudi.common.model.Defaulthoodierecordpayload hoodie.schedule.compact.only.inline 在写入操作时，是否只生成压缩计划。在hoodie.compact.inline=true时有效。 false hoodie.run.compact.only.inline 通过Sql执行run compaction命令时，是否只执行压缩操作，压缩计划不存在时直接退出。 false 父主题： 配置参考

index相关配置 参数 描述 默认值 hoodie.index.class 用户自定义索引的全路径名，索引类必须为HoodieIndex的子类，当指定该配置时，其会优先于hoodie.index.type配置。 "" hoodie.index.type 使用的索引类型，默认为布隆过滤器。可能的选项是[BLOOM | HBASE | GLOBAL_BLOOM | SIMPLE | GLOBAL_SIMPLE] 。 布隆过滤器消除了对外部系统的依赖，并存储在Parquet数据文件的页脚中。 BLOOM hoodie.index.bloom.num_entries 存储在布隆过滤器中的条目数。 假设maxParquetFileSize为128MB，averageRecordSize为1024B，因此，一个文件中的记录总数约为130K。 默认值（60000）大约是此近似值的一半。 注意： 将此值设置得太低，将产生很多误报，并且索引查找将必须扫描比其所需的更多的文件；如果将其设置得非常高，将线性增加每个数据文件的大小（每50000个条目大约4KB）。 60000 hoodie.index.bloom.fpp 根据条目数允许的错误率。 用于计算应为布隆过滤器分配多少位以及哈希函数的数量。通常将此值设置得很低（默认值：0.000000001），在磁盘空间上进行权衡以降低误报率。 0.000000001 hoodie.bloom.index.parallelism 索引查找的并行度，其中涉及Spark Shuffle。 默认情况下，根据输入的工作负载特征自动计算的。 0 hoodie.bloom.index.prune.by.ranges 为true时，从文件框定信息，可以加快索引查找的速度。 如果键具有单调递增的前缀，例如时间戳，则特别有用。 true hoodie.bloom.index.use.caching 为true时，将通过减少用于计算并行度或受影响分区的IO来缓存输入的RDD以加快索引查找。 true hoodie.bloom.index.use.treebased.filter 为true时，启用基于间隔树的文件过滤优化。与暴力模式相比，此模式可根据键范围加快文件过滤速度。 true hoodie.bloom.index.bucketized.checking 为true时，启用了桶式布隆过滤。这减少了在基于排序的布隆索引查找中看到的偏差。 true hoodie.bloom.index.keys.per.bucket 仅在启用bloomIndexBucketizedChecking并且索引类型为bloom的情况下适用。 此配置控制“存储桶”的大小，该大小可跟踪对单个文件进行的记录键检查的次数，并且是分配给执行布隆过滤器查找的每个分区的工作单位。 较高的值将分摊将布隆过滤器读取到内存的固定成本。 10000000 hoodie.bloom.index.update.partition.path 仅在索引类型为GLOBAL_BLOOM时适用。 为true时，当对一个已有记录执行包含分区路径的更新操作时，将会导致把新记录插入到新分区，而把原有记录从旧分区里删除。为false时，只对旧分区的原有记录进行更新。 true hoodie.index.hbase.zkquorum 仅在索引类型为HBASE时适用，必填选项。要连接的HBase ZK Quorum URL。 无 hoodie.index.hbase.zkport 仅在索引类型为HBASE时适用，必填选项。要连接的HBase ZK Quorum端口。 无 hoodie.index.hbase.zknode.path 仅在索引类型为HBASE时适用，必填选项。这是根znode，它将包含HBase创建及使用的所有znode。 无 hoodie.index.hbase.table 仅在索引类型为HBASE时适用，必填选项。HBase表名称，用作索引。Hudi将row_key和[partition_path, fileID, commitTime]映射存储在表中。 无 父主题： 配置参考

存储配置 参数 描述 默认值 hoodie.parquet.max.file.size Hudi写阶段生成的parquet文件的目标大小。对于DFS，这需要与基础文件系统块大小保持一致，以实现最佳性能。 120 * 1024 * 1024 byte hoodie.parquet.block.size parquet页面大小，页面是parquet文件中的读取单位，在一个块内，页面被分别压缩。 120 * 1024 * 1024 byte hoodie.parquet.compression.ratio 当Hudi尝试调整新parquet文件的大小时，预期对parquet数据进行压缩的比例。 如果bulk_insert生成的文件小于预期大小，请增加此值。 0.1 hoodie.parquet.compression.codec parquet压缩编解码方式名称，默认值为gzip。可能的选项是[gzip | snappy | uncompressed | lzo] snappy hoodie.logfile.max.size LogFile的最大值。这是在将日志文件移到下一个版本之前允许的最大值。 1GB hoodie.logfile.data.block.max.size LogFile数据块的最大值。这是允许将单个数据块附加到日志文件的最大值。 这有助于确保附加到日志文件的数据被分解为可调整大小的块，以防止发生OOM错误。此大小应大于JVM内存。 256MB hoodie.logfile.to.parquet.compression.ratio 随着记录从日志文件移动到parquet，预期会进行额外压缩的比例。 用于merge_on_read存储，以将插入内容发送到日志文件中并控制压缩parquet文件的大小。 0.35 父主题： 配置参考

同步hive表配置 参数 描述 默认值 hoodie.datasource.hive_sync.enable 是否同步hudi表信息到hive metastore。 注意： 建议该值设置为true，统一使用hive管理hudi表。 false hoodie.datasource.hive_sync.database 要同步给hive的数据库名。 default hoodie.datasource.hive_sync.table 要同步给hive的表名，建议这个值和hoodie.datasource.write.table.name保证一致。 unknown hoodie.datasource.hive_sync.username 同步hive时，指定的用户名。 hive hoodie.datasource.hive_sync.password 同步hive时，指定的密码。 hive hoodie.datasource.hive_sync.jdbcurl 连接hive jdbc指定的连接。 "" hoodie.datasource.hive_sync.use_jdbc 是否使用hive jdbc方式连接hive同步hudi表信息。建议该值设置为false，设置为false后 jdbc连接相关配置无效。 true hoodie.datasource.hive_sync.partition_fields 用于决定hive分区列。 "" hoodie.datasource.hive_sync.partition_extractor_class 用于提取hudi分区列值，将其转换成hive分区列。 org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor hoodie.datasource.hive_sync.support_timestamp 当hudi表存在timestamp类型字段时，需指定此参数为true，以实现同步timestamp类型到hive元数据中。该值默认为false，默认将timestamp类型同步为bigInt，默认情况可能导致使用sql查询包含timestamp类型字段的hudi表出现错误。 true 父主题： 配置参考

写入操作配置 表1 写入操作重要配置项 参数 描述 默认值 hoodie.datasource.write.table.name 指定写入的hudi表名。 无 hoodie.datasource.write.operation 写hudi表指定的操作类型，当前支持upsert、delete、insert、bulk_insert等方式。 upsert：更新插入混合操作 delete：删除操作 insert：插入操作 bulk_insert： 用于初始建表导入数据， 注意初始建表禁止使用upsert、insert方式 insert_overwrite：对静态分区执行insert overwrite insert_overwrite_table：动态分区执行insert overwrite，该操作并不会立刻删除全表做overwrite，会逻辑上重写hudi表的元数据，无用数据后续由hudi的clean机制清理。效率比bulk_insert + overwrite 高 upsert hoodie.datasource.write.table.type 指定hudi表类型，一旦这个表类型被指定，后续禁止修改该参数，可选值MERGE_ON_READ。 COPY_ON_WRITE hoodie.datasource.write.precombine.field 该值用于在写之前对具有相同的key的行进行合并去重。 ts hoodie.datasource.write.payload.class 在更新过程中，该类用于提供方法将要更新的记录和更新的记录做合并，该实现可插拔，如要实现自己的合并逻辑，可自行编写。 org.apache.hudi.common.model.DefaultHoodieRecordPayload hoodie.datasource.write.recordkey.field 用于指定hudi的主键，hudi表要求有唯一主键。 uuid hoodie.datasource.write.partitionpath.field 用于指定分区键，该值配合hoodie.datasource.write.keygenerator.class使用可以满足不同的分区场景。 无 hoodie.datasource.write.hive_style_partitioning 用于指定分区方式是否和hive保持一致，建议该值设置为true。 true hoodie.datasource.write.keygenerator.class 配合hoodie.datasource.write.partitionpath.field，hoodie.datasource.write.recordkey.field产生主键和分区方式。 说明： 写入设置KeyGenerator与表保存的参数值不一致时将提示需要保持一致。 org.apache.hudi.keygen.ComplexKeyGenerator 父主题： 配置参考

Cleaning Cleaning用于清理不再需要的版本数据。 Hudi使用Cleaner后台作业，不断清除不需要的旧得版本的数据。通过配置hoodie.cleaner.policy和hoodie.cleaner.commits.retained可以使用不同的清理策略和保存的commit数量。 执行cleaning有两种方式： 同步clean由参数hoodie.clean.automatic控制，默认自动开启。 关闭同步clean： datasource写入时可以通过.option("hoodie.clean.automatic", "false")来关闭自动clean。 spark-sql写入时可以通过set hoodie.clean.automatic=false;来关闭自动clean。 异步clean可以使用spark-sql来执行。 更多clean相关参数请参考compaction&cleaning配置章节。 父主题： 数据管理维护

Clustering架构 Hudi通过其写入客户端API提供了不同的操作，如insert/upsert/bulk_insert来将数据写入Hudi表。为了能够在文件大小和入湖速度之间进行权衡，Hudi提供了一个hoodie.parquet.small.file.limit配置来设置最小文件大小。用户可以将该配置设置为“0”，以强制新数据写入新的文件组，或设置为更高的值以确保新数据被“填充”到现有小的文件组中，直到达到指定大小为止，但其会增加摄取延迟。 为能够支持快速摄取的同时不影响查询性能，引入了Clustering服务来重写数据以优化Hudi 数据湖 文件的布局。 Clustering服务可以异步或同步运行，Clustering会添加了一种新的REPLACE操作类型，该操作类型将在Hudi元数据时间轴中标记Clustering操作。 Clustering服务基于Hudi的MVCC设计，允许继续插入新数据，而Clustering操作在后台运行以重新格式化数据布局，从而确保并发读写者之间的快照隔离。 总体而言Clustering分为两个部分： 调度Clustering：使用可插拔的Clustering策略创建Clustering计划。 识别符合Clustering条件的文件：根据所选的Clustering策略，调度逻辑将识别符合Clustering条件的文件。 根据特定条件对符合Clustering条件的文件进行分组。每个组的数据大小应为targetFileSize的倍数。分组是计划中定义的"策略"的一部分。此外还有一个选项可以限制组大小，以改善并行性并避免混排大量数据。 将Clustering计划以avro元数据格式保存到时间线。 执行Clustering：使用执行策略处理计划以创建新文件并替换旧文件。 读取Clustering计划，并获得ClusteringGroups，其标记了需要进行Clustering的文件组。 对于每个组使用strategyParams实例化适当的策略类（例如：sortColumns），然后应用该策略重写数据。 创建一个REPLACE提交，并更新HoodieReplaceCommitMetadata中的元数据。

将Hudi表数据同步到Hive 通过执行run_hive_sync_tool.sh可以将Hudi表数据同步到Hive中。 例如：需要将HDFS上目录为hdfs://hacluster/tmp/huditest/hudimor1_deltastreamer_partition的Hudi表同步为Hive表，表名为table hive_sync_test3，使用unite、country和state为分区键，命令示例如下： run_hive_sync_tool.sh --partitioned-by unite,country,state --base-path hdfs://hacluster/tmp/huditest/hudimor1_deltastreamer_partition --table hive_sync_test3 --partition-value-extractor org.apache.hudi.hive.MultiPartKeysValueExtractor --support-timestamp 表1 参数说明 命令 描述 必填 默认值 --database Hive database名称 N default --table Hive表名 Y - --base-file-format 文件格式 (PARQUET或HFILE) N PARQUET --user Hive用户名 N - --pass Hive密码 N - --jdbc-url Hive jdbc connect url N - --base-path 待同步的Hudi表存储路径 Y - --partitioned-by 分区键- N - --partition-value-extractor 分区类，需实现PartitionValueExtractor ，可以从HDFS路径中提取分区值 N SlashEncodedDayPartitionValueExtractor --assume-date-partitioning 以 yyyy/mm/dd进行分区从而支持向后兼容。 N false --use-pre-apache-input-format 使用com.uber.hoodie包下的InputFormat替换 org.apache.hudi包下的。除了从com.uber.hoodie迁移项目至org.apache.hudi外请勿使用。 N false --use-jdbc 使用Hive jdbc连接 N true --auto-create-database 自动创建Hive database N true --skip-ro-suffix 注册时跳过读取_ro后缀的读优化视图 N false --use-file-listing-from-metadata 从Hudi的元数据中获取文件列表 N false --verify-metadata-file-listing 根据文件系统验证Hudi元数据中的文件列表 N false --help、-h 查看帮助 N false --support-timestamp 将原始类型中'INT64'的TIMESTAMP_MICROS转换为Hive的timestamp N false --decode-partition 如果分区在写入过程中已编码，则解码分区值 N false --batch-sync-num 指定每批次同步hive的分区数 N 1000 Hive Sync时会判断表不存在时建外表并添加分区，表存在时对比表的schema是否存在差异，存在则替换，对比分区是否有新增，有则添加分区。 因此使用hive sync时有以下约束： 写入数据Schema只允许增加字段，不允许修改、删除字段。 分区目录只能新增，不会删除。 Overwrite覆写Hudi表不支持同步覆盖Hive表。 Hudi同步Hive表时，不支持使用timestamp类型作为分区列。 父主题： 写操作指导

回答 不可以，会抛HoodieKeyException异常。 Caused by: org.apache.hudi.exception.HoodieKeyException: recordKey value: "null" for field: "name" cannot be null or empty.at org.apache.hudi.keygen.SimpleKeyGenerator.getKey(SimpleKeyGenerator.java:58)at org.apache.hudi.HoodieSparkSqlWriter$$anonfun$1.apply(HoodieSparkSqlWriter.scala:104)at org.apache.hudi.HoodieSparkSqlWriter$$anonfun$1.apply(HoodieSparkSqlWriter.scala:100)

问题 Hive同步数据时报错： Caused by: java.sql.SQLException: Error while processing statement: FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. Unable to alter table. The following columns have types incompatible with the existing columns in their respective positions :__col1,__col2

问题 线程“main”报错 org.apache.kafka.common.KafkaException，构造kafka消费者失败，报错： java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set

回答 当试图从启用SSL的kafka数据源采集数据时，而安装程序无法读取jars.conf文件及其属性时，可能会发生这种情况。 要解决此问题，需要将所需的属性作为通过Spark提交的命令的一部分传递。如：--files jaas.conf,failed_tables.json --conf 'spark.driver.extraJavaOptions=-Djava.security.auth.login.config=jaas.conf' --conf 'spark.executor .extraJavaOptions=-Djava.security.auth.login.config=jaas.conf'

回答 建议在使用Hudi时，schema应该以向后兼容的方式演进。此错误通常发生在使用向后不兼容的演进方式删除某些列如“col1”后，更新parquet文件中以旧的schema写入的列“col1”，在这种情况下，parquet尝试在传入记录中查找所有当前字段，当发现“col1”不存在时，抛出上述异常。 解决这个问题的办法是使用所有schema演进版本来创建uber schema，并使用该schema作为target schema。用户可以从hive metastore中获取schema并将其与当前schema合并。

操作并发 DDL和DML中的操作，执行前，需要获取对应的锁，各操作需要获取锁的情况见表1 操作获取锁一览表，√表示需要获取该锁，一个操作仅在获取到所有需要获取的锁后，才能继续执行。 任意两个操作是否可以并发执行，可以通过如下方法确定：表1两行代表两个操作，这两行没有任意一列都标记√，即不存在某一列两行全为√。 表1 操作获取锁一览表 操作 METADATA_LOCK COMPACTION_LOCK DROP_TABLE_LOCK DELETE_SEGMENT_LOCK CLEAN_FILES_LOCK ALTER_PARTITION_LOCK UPDATE_LOCK STREAMING_LOCK CONCURRENT_LOAD_LOCK SEGMENT_LOCK CREATE TABLE - - - - - - - - - - CREATE TABLE As SELECT - - - - - - - - - - DROP TABLE √ - √ - - - - √ - - ALTER TABLE COMPACTION - √ - - - - √ - - - TABLE RENAME - - - - - - - - - - ADD COLUMNS √ √ - - - - - - - - DROP COLUMNS √ √ - - - - - - - - CHANGE DATA TYPE √ √ - - - - - - - - REFRESH TABLE - - - - - - - - - - REGISTER INDEX TABLE √ - - - - - - - - - REFRESH INDEX - √ - - - - - - - - LOAD DATA/INSERT INTO - - - - - - - - √ √ UPDATE CARBON TABLE √ √ - - - - √ - - - DELETE RECORDS from CARBON TABLE √ √ - - - - √ - - - DELETE SEGMENT by ID - - - √ √ - - - - - DELETE SEGMENT by DATE - - - √ √ - - - - - SHOW SEGMENTS - - - - - - - - - - CREATE SECONDARY INDEX √ √ - √ - - - - - - SHOW SECONDARY INDEXES - - - - - - - - - - DROP SECONDARY INDEX √ - √ - - - - - - - CLEAN FILES - - - - - - - - - - SET/RESET - - - - - - - - - - Add Hive Partition - - - - - - - - - - Drop Hive Partition √ √ √ √ √ √ - - - - Drop Partition √ √ √ √ √ √ - - - - Alter table set √ √ - - - - - - - - 父主题： CarbonData语法参考

基础操作 使用root用户登录集群客户端节点，执行如下命令： cd {客户端安装目录} source bigdata_env source Hudi/component_env kinit 创建的用户 执行hudi-cli.sh进入Hudi客户端， cd {客户端安装目录}/Hudi/hudi/bin/ ./hudi-cli.sh 即可执行各种Hudi命令，执行示例（仅部分命令，全部命令请参考Hudi官网：https://hudi.apache.org/docs/quick-start-guide/）： 查看帮助： help //查看hudi-cli的所有命令 help 'command' //查看某一个命令的帮助及参数列表。 连接表： connect --path '/tmp/huditest/test_table' 查看表信息： desc 查看compaction计划： compactions show all 查看clean计划： cleans show 执行clean： cleans run 查看commit信息： commits show 查看commit写入的分区： commit showpartitions --commit 20210127153356 20210127153356表示commit的时间戳，下同。 查看指定commit写入的文件： commit showfiles --commit 20210127153356 比较两个表的commit信息差异： commits compare --path /tmp/hudimor/mytest100 rollback指定提交（rollback每次只允许rollback最后一次commit）： commit rollback --commit 20210127164905 compaction调度： compaction schedule --hoodieConfigs 'hoodie.compaction.strategy=org.apache.hudi.table.action.compact.strategy.BoundedIOCompactionStrategy,hoodie.compaction.target.io=1,hoodie.compact.inline.max.delta.commits=1' 执行compaction compaction run --parallelism 100 --sparkMemory 1g --retry 1 --compactionInstant 20210602101315 --hoodieConfigs 'hoodie.compaction.strategy=org.apache.hudi.table.action.compact.strategy.BoundedIOCompactionStrategy,hoodie.compaction.target.io=1,hoodie.compact.inline.max.delta.commits=1' --propsFilePath hdfs://hacluster/tmp/default/tb_test_mor/.hoodie/hoodie.properties --schemaFilePath /tmp/default/tb_test_mor/.hoodie/compact_tb_base.json 创建savepoint savepoint create --commit 20210318155750 回滚指定的savepoint savepoint rollback --savepoint 20210318155750 若commit写入导致元数据冲突异常，执行commit rollback、savepoint rollback能回退数据，但不能回退Hive元数据，只能删除Hive表然后手动进行同步刷新。 commit rollback只能回退当前最新的一个commit，savepoint rollback只能回退到最新的一个savepoint。二者均不能随意指定进行回退。

回答 运行包含Reduce的Mapreduce任务时，通过-Dmapreduce.job.map.output.collector.class=org.apache.hadoop.mapred.nativetask.NativeMapOutputCollectorDelegator命令开启Native Task特性，任务在部分操作系统运行失败，日志中提示错误“version 'GLIBCXX_3.4.20' not found”。该问题原因是操作系统的GLIBCXX版本较低，导致该特性依赖的libnativetask.so.1.0.0库无法加载，进而导致任务失败。 规避手段： 设置配置项mapreduce.job.map.output.collector.class的值为org.apache.hadoop.mapred.MapTask$MapOutputBuffer。

推荐资源配置 mor表： 由于其本质上是写增量文件，调优可以直接根据hudi的数据大小（dataSize）进行调整。 dataSize如果只有几个G，推荐跑单节点运行spark，或者yarn模式但是只分配一个container。 入湖程序的并行度p设置：建议 p = （dataSize）/128M, 程序分配core的数量保持和p一致即可。内存设置建议内存大小和core的比例大于1.5:1 即一个core配1.5G内存， 堆外内存设置建议内存大小和core的比例大于0.5:1。 cow表： cow表的原理是重写原始数据，因此这种表的调优，要兼顾dataSize和最后重写的文件数量。总体来说core数量越大越好（和最后重写多少个文件数直接相关），并行度p和内存大小和mor设置类似。

Flink WebUI权限管理 访问并使用Flink WebUI进行业务操作需为用户赋予FlinkServer相关权限，Manager的admin用户没有FlinkServer的业务操作权限。 FlinkServer中应用（租户）是最大管理范围，包含集群连接管理、数据连接管理、应用管理、流表和作业管理等。 FlinkServer中有如表1所示三种资源权限： 表1 FlinkServer资源权限 权限名称 权限描述 备注 FlinkServer管理员权限 具有所有应用的编辑、查看权限。 是FlinkServer的最高权限。如果已经具有FlinkServer管理员权限，则会自动具备所有应用的权限。 应用编辑权限 具有当前应用编辑权限的用户，可以执行创建、编辑和删除集群连接、数据连接，创建流表、创建作业及运行作业等操作。 同时具有当前应用查看权限。 应用查看权限 具有当前应用查看权限的用户，可以查看应用。 - 父主题： 配置开发Flink可视化作业

示例 SHOW functions; --使用LIKE子句show functions like 'boo_%'; Function | Return Type | Argument Types | Function Type | Deterministic | Description ----------|-------------|----------------|---------------|---------------|------------- bool_and | boolean | boolean | aggregate | true | bool_or | boolean | boolean | aggregate | true | (2 rows)--如果匹配字符串中有字符与通配符冲突，可以指定转义字符来标识，示例为查询default这个schema下，table_name前缀为＂t_＂ 的所有table，转义字符为＂\＂:SHOW FUNCTIONS LIKE 'array\_%' escape '\'; Function | Return Type | Argument Types | Function Type | Deterministic | Description | Variable Arity | Built In -----------------|-------------|---------------------------------|---------------|---------------|--------------------------------------------------------------------------------------------------------|----------------|---------- array_agg | array(T) | T | aggregate | true | return an array of values | false | true array_contains | boolean | array(T), T | scalar | true | Determines whether given value exists in the array | false | true array_distinct | array(E) | array(E) | scalar | true | Remove duplicate values from the given array | false | true array_except | array(E) | array(E), array(E) | scalar | true | Returns an array of elements that are in the first array but not the second, without duplicates. | false | true array_intersect | array(E) | array(E), array(E) | scalar | true | Intersects elements of the two given arrays | false | true array_join | varchar | array(T), varchar | scalar | true | Concatenates the elements of the given array using a delimiter and an optional string to replace nulls | false | true array_join | varchar | array(T), varchar, varchar | scalar | true | Concatenates the elements of the given array using a delimiter and an optional string to replace nulls | false | true array_max | T | array(T) | scalar | true | Get maximum value of array | false | true array_min | T | array(T) | scalar | true | Get minimum value of array | false | true array_position | bigint | array(T), T | scalar | true | Returns the position of the first occurrence of the given value in array (or 0 if not found) | false | true array_remove | array(E) | array(E), E | scalar | true | Remove specified values from the given array | false | true array_sort | array(E) | array(E) | scalar | true | Sorts the given array in ascending order according to the natural ordering of its elements. | false | true array_sort | array(T) | array(T), function(T,T,integer) | scalar | true | Sorts the given array with a lambda comparator. | false | true array_union | array(E) | array(E), array(E) | scalar | true | Union elements of the two given arrays | false | true --查看所有JAVA和JDBC函数SHOW external functions; Function | Owner ----------------------------|----------- example.namespace02.repeat | admintest hetu.default.add_two | admintest (2 rows)--查看给定函数的定义信息SHOW external function example.namespace02.repeat; External Function -------------------------------------- External FUNCTION example.namespace02.repeat ( s varchar, n integer ) RETURNS varchar COMMENT '' LANGUAGE JDBC NOT DETERMINISTIC CALLED ON NULL INPUT FUNCPROPERTIES ( owner = 'admintest' )

语法 INSERT { INTO | OVERWRITE } [TABLE] table_name [(column_list)] [ PARTITION (partition_clause)] {select_statement | VALUES (value [, value ...]) [, (value [, value ...]) ...] } FROM from_statement INSERT OVERWRITE TABLE tablename1 [PARTITION (partcol1=val1, partcol2=val2 ...)] select_statement FROM from_statement INSERT INTO TABLE tablename1 [PARTITION (partcol1=val1, partcol2=val2 ...) select_statement

描述 向表中插入新的数据行。 如果指定了列名列表，那么这些列名列表必须与query语句产生列列表名完全匹配。表中不在列名列表中的每一列，其值会设置为null。 如果没有指定列名列表，则query语句产生的列必须与将要插入的列完全匹配。 使用insert into时，会往表中追加数据，而使用insert overwrite时，如果表属性“auto.purge”被设置为“true”，直接删除原表数据，再写入新的数据。 如果对象表是分区表时，insert overwrite会删除对应分区的数据而非所有数据。 insert into后面的table关键字为可选，以兼容hive语法。

示例 创建fruit和fruit_copy表： create table fruit (name varchar,price double);create table fruit_copy (name varchar,price double); 向fruit表中插入一行数据： insert into fruit values('LIchee',32);-- 兼容写法示例,带上table关键字insert into table fruit values('Cherry',88); 向fruit表中插入多行数据： insert into fruit values('banana',10),('peach',6),('lemon',12),('apple',7); 将fruit表中的数据行加载到fruit_copy表中，执行后表中有5条记录： insert into fruit_copy select * from fruit; 先清空fruit_copy表，再将fruit中的数据加载到表中，执行之后表中有2条记录： insert overwrite fruit_copy select * from fruit limit 2; 对于varchar类型，仅当目标表定义的列字段长度大于源表的实际字段长度时，才可以使用INSERT... SELECT...的形式从源表中查数据并且插入到目标表： create table varchar50(c1 varchar(50));insert into varchar50 values('hetuEngine');create table varchar100(c1 varchar(100));insert into varchar100 select * from varchar50; 分区表使用insert overwrite语句时，只会清理插入值所在分区的数据，而不是整个表： --创建表create table test_part (id int, alias varchar) partitioned by (dept_id int, status varchar);insert into test_part partition(dept_id=10, status='good') values (1, 'xyz'), (2, 'abc');select * from test_part order by id; id | alias | dept_id | status ----|-------|---------|-------- 1 | xyz | 10 | good 2 | abc | 10 | good (2 rows)--清理分区partition(dept_id=25, status='overwrite')，并插入一条数据insert overwrite test_part (id, alias, dept_id, status) values (3, 'uvw', 25, 'overwrite'); select * from test_part ; id | alias | dept_id | status ----|-------|---------|----------- 1 | xyz | 10 | good 2 | abc | 10 | good 3 | uvw | 25 | overwrite--清理分区partition(dept_id=10, status='good')，并插入一条数据insert overwrite test_part (id, alias, dept_id, status) values (4, 'new', 10, 'good');select * from test_part ordr; id | alias | dept_id | status ----|-------|---------|----------- 3 | uvw | 25 | overwrite 4 | new | 10 | good (2 rows)--分区表插入数据create table test_p_1(name string, age int) partitioned by (provice string, city string); create table test_p_2(name string, age int) partitioned by (provice string, city string); -- 填充数据到test_p_1 insert into test_p_1 partition (provice = 'hebei', city= 'baoding') values ('xiaobei',15),( 'xiaoming',22);-- 根据test_p_1 插入数据到test_p_2-- 方式一from test_p_1 insert into table test_p_2 partition (provice = 'hebei', city= 'baoding') select name,age; -- 方式二insert into test_p_2 partition(provice = 'hebei', city= 'baoding') select name,age from test_p_1;

示例 --查看show_table1的所有表属性SHOW TBLPROPERTIES show_table1;SHOW TBLPROPERTIES --------------------------------------------------------------------------- STATS_GENERATED_VIA_STATS_TASK workaround for potential lack of HIVE-12730 numFiles 0 numRows 0 orc.compress.size 262144 orc.compression.codec GZIP orc.row.index.stride 10000 orc.stripe.size 67108864 presto_query_id 20210322_094034_00306_s4s9e@default@HetuEngine presto_version rawDataSize 0 totalSize 0 transient_lastDdlTime 1616406034 (1 row) --查看show_table1的压缩算法SHOW TBLPROPERTIES show_table1('orc.compression.codec');SHOW TBLPROPERTIES --------------------- GZIP (1 row)

参数说明 IN | FROM schema_name 指定schema名称，未指定时默认使用当前的schema。 LIKE 'identifier_with_wildcards' identifier_with_wildcards只支持包含“*”和“|”的规则匹配表达式。 其中“*”可以匹配单个或多个字符，“|”适用于匹配多种规则匹配表达式中的任意一种的情况，它用于分隔这些规则匹配表达式。 规则匹配表达式首尾的空格，不会参与匹配计算。 partition_spec 一个可选参数，使用键值对来指定分区列表，键值对之间通过逗号分隔。需要注意，指定分区时，表名不支持模糊匹配。

示例 显示能够创建orders表的SQL 语句： show create table orders; Create Table ---------------------------------------- CREATE TABLE hive.default.orders ( orderkey bigint, orderstatus varchar, totalprice double, orderdate date ) COMMENT 'tt1_tab' WITH ( external = true, format = 'ORC', location = 'hdfs://hacluster/user', orc_compress = 'GZIP', orc_compress_size = 262144, orc_row_index_stride = 10000, orc_stripe_size = 67108864 ) (1 row)