华为云用户手册

查询Hudi的Schema演进表对应的Hive外部表 如果该Hudi表为Schema演进表（表的字段执行过修改），则在Hive客户端查询该表时还需额外设置一个参数： set hive.exec.schema.evolution=true; 例如以cow表实时视图的查询举例，其他各个视图的查询都要额外添加该参数： set hive.exec.schema.evolution=true; select * from hudicow;

Hudi表对应的Hive外部表介绍 Hudi源表对应一份HDFS的数据，通过Spark组件、Flink组件或者Hudi客户端，可以将Hudi表的数据映射为Hive外部表，基于该外部表，Hive可以进行实时视图查询、读优化视图查询以及增量视图查询。 根据Hudi源表的类型的不同，提供不同的视图查询： Hudi源表类型为Copy On Write时，可以映射为Hive的一张外部表，该表可以提供实时视图查询以及增量视图查询。 Hudi源表类型为Merge On Read时，可以映射为Hive的两张外部表（ro表和rt表），ro表提供读优化视图查询，rt表提供实时视图查询以及增量视图查询。 不能对Hudi表映射的Hive外部表做增删改操作（即insert、update、delete、load、merge、alter、msck），只支持查询操作（select）。 表授权：不支持修改类权限（update、Alter、write、All）。 备份与恢复：由于ro表和rt表均由同一个Hudi源表映射的，备份其中一张表，另一张也会跟着备份，恢复也是同时恢复的，因此只需备份其中一张表即可。 组件版本： Hive： FusionInsight _HD_xxx，Hive内核版本3.1.0。 Spark2x：FusionInsight_Spark2x_xxx，Hudi内核版本：0.11.0。

浮点函数 infinity() → double 返回表示正无穷大的常数。 select infinity();-- Infinity is_finite(x) → boolean 判断x是否有限值。 select is_finite(infinity());-- false select is_finite(50000);--true is_infinite(x) → boolean 判断x是否无穷大。 select is_infinite(infinity());-- true select is_infinite(50000);--false is_nan(x) → boolean 判断x是否非数字。 --输入的值必须为double类型 select is_nan(null); -- NULL select is_nan(nan()); -- true select is_nan(45);-- false nan() → double 返回表示非数字的常数。 select nan(); -- NaN

三角函数 所有三角函数的参数都是以弧度表示。参考单位转换函数degrees()和radians()。 acos(x) → double 求反余弦值。 SELECT acos(-1);-- 3.14159265358979 asin(x) → double 求反正弦值。 SELECT asin(0.5);-- 0.5235987755982989 atan(x) → double 求x的反正切值。 SELECT atan(1);-- 0.7853981633974483 atan2(y, x) → double 返回y/x的反正切值。 SELECT atan2(2,1);-- 1.1071487177940904 cos(x) → double 返回x的余弦值。 SELECT cos(-3.1415927);-- -0.9999999999999989 cosh(x) → double 返回x的双曲余弦值。 SELECT cosh(3.1415967);-- 11.592000006553231 sin(x) → double 求x的正弦值。 SELECT sin(1.57079);-- 0.9999999999799858 tan(x) → double 求x的正切值。 SELECT tan(20);-- 2.23716094422474 tanh(x) → double 求x双曲正切值。 select tanh(3.1415927);-- 0.9962720765661324

单表并发控制配置 表6 单表并发控制参数配置 参数 描述 默认值 hoodie.write.lock.provider 指定lock provider，不建议使用默认值，使用org.apache.hudi.hive.HiveMetastoreBasedLockProvider。 org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider hoodie.write.lock.hivemetastore.database Hive的database。 无 hoodie.write.lock.hivemetastore.table Hive的table name。 无 hoodie.write.lock.client.num_retries 重试次数。 10 hoodie.write.lock.client.wait_time_ms_between_retry 重试间隔。 10000 hoodie.write.lock.conflict.resolution.strategy lock provider类，必须是ConflictResolutionStrategy的子类。 org.apache.hudi.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy hoodie.write.lock.zookeeper.base_path 存放ZNodes的路径，同一张表的并发写入需配置一致。 无 hoodie.write.lock.zookeeper.lock_key ZNode的名称，建议与Hudi表名相同。 无 hoodie.write.lock.zookeeper.connection_timeout_ms ZooKeeper连接超时时间。 15000 hoodie.write.lock.zookeeper.port ZooKeeper端口号。 无 hoodie.write.lock.zookeeper.url ZooKeeper的url。 无 hoodie.write.lock.zookeeper.session_timeout_ms ZooKeeper的session过期时间。 60000

Clustering配置 本章节内容仅使用于 MRS 3.2.0及之后版本。 Clustering中有两个策略分别是hoodie.clustering.plan.strategy.class和hoodie.clustering.execution.strategy.class。一般情况下指定plan.strategy为SparkRecentDaysClusteringPlanStrategy或者SparkSizeBasedClusteringPlanStrategy时，execution.strategy不需要指定。但当plan.strategy为SparkSingleFileSortPlanStrategy时，需要指定execution.strategy为SparkSingleFileSortExecutionStrategy。 表7 Clustering参数配置 参数 描述 默认值 hoodie.clustering.inline 是否同步执行clustering。 false hoodie.clustering.inline.max.commits 触发clustering的commit数。 4 hoodie.clustering.async.enabled 是否启用异步执行clustering。 说明： 此参数仅适用于MRS 3.3.0-LTS及之后版本。 false hoodie.clustering.async.max.commits 异步执行时触发clustering的commit数。 说明： 此参数仅适用于MRS 3.3.0-LTS及之后版本。 4 hoodie.clustering.plan.strategy.target.file.max.bytes 指定clustering后每个文件大小最大值。 1024 * 1024 * 1024 byte hoodie.clustering.plan.strategy.small.file.limit 小于该大小的文件会被clustering。 300 * 1024 * 1024 byte hoodie.clustering.plan.strategy.sort.columns clustering用以排序的列。 无 hoodie.layout.optimize.strategy Clustering执行策略，可选linear、z-order、hilbert三种排序方式。 linear hoodie.layout.optimize.enable 使用z-order、hilbert时需要开启。 false hoodie.clustering.plan.strategy.class 筛选FileGroup进行clustering的策略类，默认筛选小于hoodie.clustering.plan.strategy.small.file.limit阈值的文件。 org.apache.hudi.client.clustering.plan.strategy.SparkSizeBasedClusteringPlanStrategy hoodie.clustering.execution.strategy.class 执行clustering的策略类（RunClusteringStrategy的子类），用以定义群集计划的执行方式。 默认类们按指定的列对计划中的文件组进行排序，同时满足配置的目标文件大小。 org.apache.hudi.client.clustering.run.strategy.SparkSortAndSizeExecutionStrategy hoodie.clustering.plan.strategy.max.num.groups 设置执行clustering时最多选择多少个FileGroup，该值越大并发度越大。 30 hoodie.clustering.plan.strategy.max.bytes.per.group 设置执行clustering时每个FileGroup最多有多少数据参与clustering。 2 * 1024 * 1024 * 1024 byte

存储配置 表4 存储参数配置 参数 描述 默认值 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

compaction&cleaning配置 表5 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相关配置 表3 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]映射存储在表中。 无

同步Hive表配置 表2 同步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的行进行合并去重。 指定为具体的表字段 hoodie.datasource.write.payload.class 在更新过程中，该类用于提供方法将要更新的记录和更新的记录做合并，该实现可插拔，如要实现自己的合并逻辑，可自行编写。 org.apache.hudi.common.model.DefaultHoodieRecordPayload hoodie.datasource.write.recordkey.field 用于指定Hudi的主键，Hudi表要求有唯一主键。 指定为具体的表字段 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

配置参数 在Spark客户端的“spark-defaults.conf”配置文件中进行设置。 参数 说明 默认值 spark.sql.streaming.stateStore.providerClass 用于管理有状态流查询中的状态数据的类。此类必须是StateStoreProvider的子类，并且必须具有零参数构造函数。 配置参数值为org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider即可选择RocksDB作为状态后端。 org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider

操作场景 对于Spark应用来说，资源是影响Spark应用执行效率的一个重要因素。当一个长期运行的服务（比如JD BCS erver），如果分配给它多个Executor，可是却没有任何任务分配给它，而此时有其他的应用却资源紧张，这就造成了很大的资源浪费和资源不合理的调度。 动态资源调度就是为了解决这种场景，根据当前应用任务的负载情况，实时的增减Executor个数，从而实现动态分配资源，使整个Spark系统更加健康。

聚合模型示例 聚合列不支持修改聚合类型。 在col1列后添加new_col列（key列）： ALTER TABLE example_db.my_table ADD COLUMN new_col INT DEFAULT "0" AFTER col1; 在col1后添加new_col列（Value列SUM聚合类型）： ALTER TABLE example_db.my_table ADD COLUMN new_col INT SUM DEFAULT "0" AFTER col1; 修改col1列的类型为BIGINT（Key列）： ALTER TABLE example_db.my_table MODIFY COLUMN col1 BIGINT DEFAULT "1"; 修改col1列的类型为BIGINT（Value列）： ALTER TABLE example_db.my_table MODIFY COLUMN col1 BIGINT MAX DEFAULT "1"; 删除col1列： ALTER TABLE example_db.my_table DROP COLUMN col1;

非聚合模型示例 在col1列后添加new_col列（添加Key列）： ALTER TABLE example_db.my_table ADD COLUMN new_col INT KEY DEFAULT "0" AFTER col1; 在col1列后添加new_col列（添加Value列）： ALTER TABLE example_db.my_table ADD COLUMN new_col INT DEFAULT "0" AFTER col1; 修改col1列的类型为BIGINT（Key列）： ALTER TABLE example_db.my_table MODIFY COLUMN col1 BIGINT KEY DEFAULT "1"; 修改col1列的类型为BIGINT（Value列）： ALTER TABLE example_db.my_table MODIFY COLUMN col1 BIGINT DEFAULT "1"; 删除col1列： ALTER TABLE example_db.my_table DROP COLUMN col1;

使用HetuEngine UDF 使用客户端访问： 进入HetuEngine客户端。 执行如下命令创建HetuEngine UDF： CREATE FUNCTION example.namespace01.add_two ( num integer ) RETURNS integer LANGUAGE JAVA DETERMINISTIC SYMBOL "com.xxx.bigdata.hetuengine.functions.AddTwo" URI "hdfs://hacluster/udf/hetuserver/udf-test-0.0.1-SNAPSHOT.jar"; 执行如下命令使用HetuEngine UDF: select example.namespace01.add_two(2); _col0 ------- 4 (1 row) 函数实现类中通过重载方法来区分同名的不同函数，在创建HetuEngine UDF时要指定不同的函数名称。

参数说明 表1 算子参数说明 参数 含义 类型 是否必填 默认值 输出分隔符 配置分隔符。 string 是 无 换行符 用户根据数据实际情况，填写字符串作为换行符。支持任何字符串。默认使用操作系统的换行符。 string 否 \n 输出字段 配置输出信息： 位置：配置输出字段的位置。 字段名：配置输出字段的字段名。 类型：配置字段类型，字段类型为“DATE”或“TIME”或“TimeStamp”时，需指定特定时间格式，其他类型指定无效。时间格式如：“yyyyMMdd HH:mm:ss”。 长度：配置字段长度，字段值实际长度太长则按配置的长度截取，“类型”为“CHAR”时实际长度不足则空格补齐，“类型”为“VARCHAR”时实际长度不足则不补齐。 map 否 无

配置场景 文件压缩可以减少储存文件的空间，并且提高数据从磁盘读取和网络传输的速度。HDFS有Gzip和Snappy这两种默认压缩格式。本章节为HDFS新增加的压缩格式LZC（Lempel-Ziv Compression）提供配置方法。这种压缩格式增强了Hadoop压缩能力。有关Snappy的详细信息，请参阅http://code.google.com/p/snappy/。 本章节适用于MRS 3.x及后续版本。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“HiveServer2 脚本”按钮，将其拖到操作区中。 在弹出的“HiveServer2 Script”窗口中配置HDFS上的脚本路径，例如“/user/admin/examples/apps/hive2/script.q”，然后单击“添加”。 单击“参数+”，添加输入输出参数。 例如输入参数为“INPUT=/user/admin/examples/input-data/table”，输出参数为“OUTPUT=/user/admin/examples/output-data/hive2_workflow”。 单击右上角的配置按钮。在打开的配置界面中，单击“删除+”，添加删除目录，例如“/user/admin/examples/output-data/hive2_workflow”。 配置“作业 XML”，值为“客户端安装目录/Oozie/oozie-client-*/examples/apps/hive/hive-site.xml”上传至HDFS目录中所在路径，例如“/user/admin/examples/apps/hive2/hive-site.xml”。HiveServer2 URL”及其他参数无需配置。 如果以上的参数和值在使用过程中发生了修改，可在“Oozie客户端安装目录/oozie-client-*/conf/hive-site.xml”文件中查询。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Hive2-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

使用BulkLoad工具向HBase迁移数据 HBase的数据都是存储在HDFS中的，数据导入即是加载存放在HDFS中的数据到HBase表中。Apache HBase提供了“Import”和“ImportTsv”工具用于批量导入HBase数据。 “Import”通过“org.apache.hadoop.hbase.mapreduce.Import”方法导入已导出至HDFS中的HBase数据。 “ImportTsv”通过“org.apache.hadoop.hbase.mapreduce.ImportTsv”可将TSV格式的数据加载到HBase中。 更多详细信息请参见：http://hbase.apache.org/2.2/book.html#tools。 父主题： 使用HBase

ClickHouse数据类型说明 MRS的ClickHouse服务数据类型如表1所示。 ClickHouse完整数据类型介绍，请参考开源官方数据类型介绍。 表1 ClickHouse数据类型 分类 关键字 数据类型 描述 数据类型 Int8 Int8 取值范围：-128~127 Int16 Int16 取值范围：-32768~32767 Int32 Int32 取值范围：-2147483648~2147483647 Int64 Int64 取值范围：-9223372036854775808~9223372036854775807 浮点类型 Float32 单精度浮点数 同C语言Float类型，单精度浮点数在机内占4个字节，用32位二进制描述。 Float64 双精度浮点数 同C语言Double类型，双精度浮点数在机内占8个字节，用64位二进制描述。 Decimal类型 Decimal Decimal 有符号的定点数，可在加、减和乘法运算过程中保持精度。支持几种写法： Decimal(P, S) Decimal32(S) Decimal64(S) Decimal128(S) 说明： P：精度，有效范围：[1:38]，决定可以有多少个十进制数字（包括分数）。 S：规模，有效范围：[0：P]，决定数字的小数部分中包含的小数位数。 字符串类型 String 字符串 字符串可以是任意长度的。它可以包含任意的字节集，包含空字节。因此，字符串类型可以代替其他DBMSs中的VARCHAR、BLOB、CLOB等类型。 FixedString 固定字符串 当数据的长度恰好为N个字节时，FixedString类型是高效的。在其他情况下，这可能会降低效率。可以有效存储在FixedString类型的列中的值的示例： 二进制表示的IP地址 语言代码（ru_RU, en_US … ） 货币代码（RUB … ） 二进制表示的哈希值（MD5使用FixedString（16），SHA256使用FixedString（32）） 时间日期类型 Date 日期 用两个字节存储，表示从1970-01-01（无符号）到当前的日期值。日期中没有存储时区信息。 DateTime 时间戳 用四个字节（无符号的）存储Unix时间戳。允许存储与日期类型相同的范围内的值。最小值为1970-01-01 00:00:00。时间戳类型值精确到秒（没有闰秒）。时区使用启动客户端或服务器时的系统时区。 DateTime64 DateTime64 此类型允许以日期（date）加时间（time）的形式来存储一个时刻的时间值。 布尔型 Boolean Boolean ClickHouse没有单独的类型来存储布尔值。可以使用UInt8类型，取值限制为0或1。 数组类型 Array Array Array(T)，由T类型元素组成的数组。T可以是任意类型，包含数组类型。但不推荐使用多维数组，ClickHouse对多维数组的支持有限。例如，不能在MergeTree表中存储多维数组。 元组类型 Tuple Tuple Tuple(T1, T2, ...)，元组，其中每个元素都有单独的类型，不能在表中存储元组（除了内存表）。它们可以用于临时列分组。在查询中，IN表达式和带特定参数的lambda函数可以来对临时列进行分组。 Domains数据类型 Domains Domains Domains类型是特定实现的类型： IPv4是与UInt32类型保持二进制兼容的Domains类型，用于存储IPv4地址的值。它提供了更为紧凑的二进制存储的同时支持识别可读性更加友好的输入输出格式。 枚举类型 Enum8 Enum8 取值范围：-128~127 Enum保存'string' = integer的对应关系，例如：Enum8('hello' = 1, 'world' = 2) Enum16 Enum16 取值范围：-32768~32767 可为空 Nullable Nullable 除非在ClickHouse服务器配置中另有说明，否则NULL是任何Nullable类型的默认值。Nullable类型字段不能包含在表索引中。 可以与TypeName的正常值存放一起。例如，Nullable(Int8)类型的列可以存储Int8类型值，而没有值的行将存储NULL。 嵌套类型 nested nested 嵌套的数据结构就像单元格内的表格。嵌套数据结构的参数（列名和类型）的指定方式与CREATE TABLE查询中的指定方式相同。每个表行都可以对应于嵌套数据结构中的任意数量的行。 示例：Nested(Name1 Type1, Name2 Type2, …)

前提条件 已安装ClickHouse客户端，例如客户端安装目录为“/opt/client”。 如果集群为安全模式需要创建一个具有ClickHouse相关权限的用户，例如创建用户“clickhouseuser”，具体请参考创建ClickHouse角色。 准备待导入的数据文件，并将数据文件上传到客户端节点目录，例如上传到目录“/opt/data”。ClickHouse支持的所有数据类型请参考：https://clickhouse.com/docs/en/interfaces/formats

操作步骤 以客户端安装用户，登录客户端所在节点。 进入多线程写入工具“clickhouse_insert_tool”所在目录： cd /opt/client/ClickHouse/clickhouse_insert_tool 使用文本编辑器打开clickhouse_insert_tool.sh，按照注释填写所需信息： 参数 参数描述 示例 datapath 待导入数据所在目录。 /opt/data balancer_ip_list ClickHouse服务Balancer实例IP地址列表，整体使用括号括起，单个IP使用双引号引起，IP之间使用空格分隔。 ("192.168.1.1" "192.168.1.2") balancer_tcp_port ClickHouse服务Balancer实例TCP端口。 21428 local_table_name 待导入的本地库名.本地表名。 testdb1.testtb1 thread_num 并发导入线程数。 10 data_format 待导入数据的格式。 CS V is_security_cluster 是否为安全模式集群。 true：表示安全模式 false：表示普通模式 true 保存修改后的 “clickhouse_insert_tool.sh”，并执行如下命： cd /opt/client source bigdata_env 安全模式（开启Kerberos认证）执行kinit命令，普通模式（关闭Kerberos认证）无需执行： kinit clickhouseuser 运行脚本，导入数据。 ./ClickHouse/clickhouse_insert_tool/clickhouse_insert_tool.sh 登录ClickHouse客户端节点，链接服务端，具体请参考ClickHouse客户端使用实践。 执行如下命令，查询插入数据的本地表对应的分布式表，查看结果是否符合预期： select count(1) from testdb1.testtb1_all;

工具介绍 在Hadoop大规模生产集群中，由于HDFS的元数据都保存在NameNode的内存中，集群规模受制于NameNode单点的内存限制。如果HDFS中有大量的小文件，会消耗NameNode大量内存，还会大幅降低读写性能，延长作业运行时间。因此，小文件问题是制约Hadoop集群规模扩展的关键问题。 本工具主要有如下两个功能： 扫描表中有多少低于用户设定阈值的小文件，返回该表目录中所有数据文件的平均大小。 对表文件提供合并功能，用户可设置合并后的平均文件大小。

Web安全 Flink Web安全加固，支持白名单过滤，Flink Web只能通过YARN代理访问，支持安全头域增强。在Flink集群中，各部件的监测端口支持范围可配置。 编码规范： 说明：Web Service客户端和服务器间使用相同的编码方式，是为了防止出现乱码现象，也是实施输入校验的基础。 安全加固：web server响应消息统一采用UTF-8字符编码。 支持IP白名单过滤： 说明：防止非法用户登录，需在web server侧添加IP Filter过滤源IP非法的请求。 安全加固：支持IP Filter实现Web白名单配置，配置项是“jobmanager.web.allow-access-address”，默认情况下只支持YARN用户接入。 安装客户端之后需要将客户端节点IP追加到jobmanager.web.allow-access-address配置项中。 禁止将文件绝对路径发送到客户端： 说明：文件绝对路径发送到客户端会暴露服务端的目录结构信息，有助于攻击者遍历了解系统，为攻击者攻击提供帮助。 安全加固：Flink配置文件中所有配置项中如果包含以/开头的，则删掉第一级目录。 同源策略： 说明：如果两个URL的协议，主机和端口均相同，则它们同源；如果不同源，默认不能相互访问；除非被访问者在其服务端显示指定访问者的来源。 安全加固：响应头“Access-Control-Allow-Origin”头域默认配置为YARN集群ResourceManager的IP地址，如果源不是来自YARN的，则不能互相访问。 防范敏感信息泄露： 说明：带有敏感数据的Web页面都应该禁止缓存，以防止敏感信息泄漏或通过代理服务器上网的用户数据互窜现象。 安全加固：添加“Cache-control”、“Pragma”、“Expires”安全头域，默认值为：“Cache-Control：no-store”，“Pragma ：no-cache”，“Expires ： 0”。实现了安全加固，Flink和web server交互的内容将不会被缓存。 防止劫持： 说明：由于点击劫持（ClickJacking）和框架盗链都利用到框架技术，所以需要采用安全措施。 安全加固：添加“X-Frame-Options”安全头域，给浏览器提供允许一个页面可否在“iframe”、“frame”或“object”网站中的展现页面的指示，如果默认配置为“X-Frame-Options: DENY”，则确保任何页面都不能被嵌入到别的“iframe”、“frame”或“object”网站中，从而避免了点击劫持 (clickjacking) 的攻击。 对Web Service接口调用记录日志： 说明：对“Flink webmonitor restful”接口调用进行日志记录。 安全加固：“access log”支持配置：“jobmanager.web.accesslog.enable”，默认为“true”。且日志保存在单独的“webaccess.log”文件中。 跨站请求（CSRF）伪造防范： 说明：在B/S应用中，对于涉及服务器端数据改动（如增加、修改、删除）的操作必须进行跨站请求伪造的防范。跨站请求伪造是一种挟制终端用户在当前已登录的Web应用程序上执行非本意的操作的攻击方法。 安全加固：现有请求修改的接口有2个post，1个delete，其余均是get请求，非get请求的接口均已删除。 异常处理： 说明：应用程序出现异常时，捕获异常，过滤返回给客户端的信息，并在日志中记录详细的错误信息。 安全加固：默认的错误提示页面，进行信息过滤，并在日志中记录详细的错误信息。新加四个配置项，默认配置为FusionInsight提供的跳转URL，错误提示页面跳转到固定配置的URL中，防止暴露不必要的信息。 表1 四个配置项参数介绍 参数 描述 是否必选配置 jobmanager.web.403-redirect-url web403页面，访问如果遇到403错误，则会重定向到配置的页面。 是 jobmanager.web.404-redirect-url web404页面，访问如果遇到404错误，则会重定向到配置的页面。 是 jobmanager.web.415-redirect-url web415页面，访问如果遇到415错误，则会重定向到配置的页面。 是 jobmanager.web.500-redirect-url web500页面，访问如果遇到500错误，则会重定向到配置的页面。 是 HTML5安全： 说明：HTML5是下一代的Web开发规范，为开发者提供了许多新的功能并扩展了标签。这些新的标签及功能增加了攻击面，存在被攻击的风险（例如跨域资源共享、客户端存储、WebWorker、WebRTC、WebSocket等）。 安全加固：添加“Access-Control-Allow-Origin”配置，如运用到跨域资源共享功能，可对HTTP响应头的“Access-Control-Allow-Origin”属性进行控制。 Flink不涉及如客户端存储、WebWorker、WebRTC、WebSocket等安全风险。

Flink认证和加密 Flink集群中，各部件支持认证。 Flink集群内部各部件和外部部件之间，支持和外部部件如YARN、HDFS、ZooKeeprer进行kerberors认证。 Flink集群内部各部件之间，如Flink client和JobManager、JobManager和TaskManager、TaskManager和TaskManager之间支持security cookie认证。 Flink集群中，各部件支持SSL加密传输；集群内部各部件之间，如Flink client和JobManager、JobManager和TaskManager、TaskManager和TaskManager之间支持SSL加密传输。

ACL控制 在HA模式下，支持ACL控制。 Flink在HA模式下，支持用ZooKeeper来管理集群和发现服务。ZooKeeper支持SASL ACL控制，即只有通过SASL（kerberos）认证的用户，才有往ZK上操作文件的权限。如果要在Flink上使用SASL ACL控制，需要在Flink配置文件中设置如下配置： high-availability.zookeeper.client.acl: creator zookeeper.sasl.disable: false 具体配置项介绍请参考HA。

回答 Kafka重启成功后应用会按照batch时间把2017/05/11 10:57:00~2017/05/11 10:58:00缺失的RDD补上（如图2所示），尽管UI界面上显示读取的数据个数为“0”，但实际上这部分数据在补的RDD中进行了处理，因此，不存在数据丢失。 Kafka重启时间段的数据处理机制如下。 Spark Streaming应用使用了state函数（例如：updateStateByKey），在Kafka重启成功后，Spark Streaming应用生成2017/05/11 10:58:00 batch任务时，会按照batch时间把2017/05/11 10:57:00~2017/05/11 10:58:00缺失的RDD补上（Kafka重启前Kafka上未读取完的数据，属于2017/05/11 10:57:00之前的batch），如图2所示。 图2 重启时间段缺失数据处理机制

问题 在Spark Streaming应用执行过程中重启Kafka时，应用无法从Kafka获取topic offset，从而导致生成Job失败。如图1所示，其中2017/05/11 10:57:00~2017/05/11 10:58:00为Kafka重启时间段。2017/05/11 10:58:00重启成功后对应的“Input Size”的值显示为“0 records”。 图1 Web UI界面部分batch time对应Input Size为0 records

配置场景 本章节适用于MRS 3.x及后续版本。 数个成品Hadoop集群由于NameNode超负荷运行并失去响应而发生故障。 这种阻塞现象是由于Hadoop的初始设计造成的。在Hadoop中，NameNode作为单独的机器，在其namespace内协调HDFS的各种操作。这些操作包括获取数据块位置，列出目录及创建文件。NameNode接受HDFS的操作，将其视作RPC调用并置入FIFO调用队列，供读取线程处理。虽然FIFO在先到先服务的情况下足够公平，但如果用户执行的I/O操作较多，相比I/O操作较少的用户，将获得更多的服务。在这种情况下，FIFO有失公平并且会导致延迟增加。 图1 基于FIFO调用队列的NameNode请求处理 如果将FIFO队列替换为一种被称作FairCallQueue的新型队列，这种情况就能够得到改善。按照这种方法，FAIR队列会根据调用者的调用规模将传入的RPC调用分配至多个队列中。调度模块会跟踪最新的调用，并为调用量较小的用户分配更高的优先级。 图2 基于FAIRCallQueue的NameNode请求处理