KS3-HDFS实践

环境准备

1. 登录KS3控制台，联系客服开通KS3-HDFS服务。

2. 记录KS3-HDFS的Bucket域名。

3. 大数据集群安装KS3-HDFS对应的Jar包。【Jar包地址请工单联系获取】

rz -bye

注：如果之前有使用hadoop-KS3直连客户，需要先将hadoop-ks3的Jar包删除。

4. 获取KMR ip并将Jar包拷贝到每个节点下。

awk '/clusterid/{print $NF}' /etc/hosts > kmrip
pscp.pssh -h kmrip alluxio-shaded-client-2.9.0.2-SNAPSHOT.jar /root

注：以上命令中，clusterid为变量值，填写对应的集群id。

5. 将Jar包移动至当前节点的hadoop/lib目录下。

mv alluxio-shaded-client-2.9.0.2-SNAPSHOT.jar /mnt/kmr/hadoop/1/hadoop-3.1.1/lib/

注：Jar包需要上传到集群下每个节点的hadoop/lib目录下。

6. 修改以下文件内容，增加相关KS3配置。

• core-site.xml【更改后需要重启Hdfs和Hive服务生效】

alluxio.user.client.base.filesystem.class=alluxio.client.file.Ks3HdfsMetaFileSystem
fs.ks3.impl=alluxio.hadoop.Ks3HdfsFileSystem
alluxio.master.rpc.addresses=ks3 hdfs endpoint地址
alluxio.ks3.hdfs.mode=true

注：

• 用户需填写对应的ks3-hdfs endpoint，示例：cn-beijing-internal.ks3-dls.ksyuncs.com。
• 如使用金山云KMR产品，可通过金山云控制台进行配置操作，注意所有的配置范围均须配置。

• mapred-site.xml 【更改后需要重启MapReduce】

mapred.output.committer.class=org.apache.hadoop.mapred.FileOutputCommitter
mapreduce.outputcommitter.factory.scheme.ks3=org.apache.hadoop.mapreduce.lib.output.PathOutputCommitterFactory

• spark-defaults 【更改后需要重启Spark】

spark.driver.extraClassPath=/mnt/kmr/hadoop/1/hadoop-3.1.1/lib
spark.executor.extraClassPath=/mnt/kmr/hadoop/1/hadoop-3.1.1/lib
spark.sql.parquet.output.committer.class=org.apache.parquet.hadoop.ParquetOutputCommitter

• hbase-site.xml 【更改后需要重启Hbase服务】

hbase.rootdir=ks3://${bucket-name}/hbase  #将bucket name替换为开通ks3-hdfs的桶名称
hbase.wal.dir=hdfs://hdfs-ha/hbase
hbase.bulkload.staging.dir=/tmp/hbase-staging

7. 验证。

hdfs dfs -ls ks3://${bucket}/

注：以上命令用于验证是否配置成功，如果list出Bucket下目录/文件时，代表配置成功。

8. 切换Hadoop身份。

su - hadoop

任务执行

以下将介绍KMR Hadoop、Hive或Spark如何操作KS3-HDFS。

相关信息

• 访问路径： ks3://${bucket}/path
• KS3-HDFS服务会通过您的角色授权访问KS3数据源进行所需操作

Hadoop 对接KS3

本示例为您展示如何在KS3-HDFS上进行hadoop wordcount作业。

hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.1.jar wordcount ks3://${bucket}/input-path ks3://${bucket}/output-path

注：根据实际情况填写对应的输入地址（input-path）和输出地址（output-path），输出地址对应的文件名需要在KS3 Bucket中不存在。

Spark 对接KS3

本示例为您展示，Spark如何读取KS3中数据，并将处理完的数据写回至KS3。

import org.apache.spark.{SparkConf, SparkContext}
val conf = new SparkConf().setAppName("Test ks3")
val sc = new SparkContext(conf)
val pathIn = "ks3://${bucket}/path/*/*"
val inputData = sc.textFile(pathIn)
val cnt = inputData.count
println(s"count: $cnt")
val outputPath = "ks3://${bucket}/path"
val outpuData = inputData.map(e => s"$e has been processed.")
outpuData.saveAsTextFile(outputPath)

PySpark接入KS3示例

本示例为您展示，PySpark如何访问KS3中数据，并将处理完的数据写回至KS3。

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Python Spark SQL ks3 example").getOrCreate()
pathIn = "ks3://${bucket}/path/to/read"
df = spark.read.text(pathIn)
cnt = df.count()
print(cnt)
outputPath = "ks3://${bucket}/path/to/write"
df.write.format("parquet").mode('overwrite').save(outputPath)

Hive 对接KS3

以下将介绍如何在KMR集群中开发Hive作业流程。

在Hive中使用KS3

修改完core-site.xml配置文件后，需要重启Hive服务生效。
在Hive中读写KS3时，先创建一个external的表。

CREATE EXTERNAL TABLE eusers (
  userid INT) 
 LOCATION 'ks3://${bucket}/path';

注：数据和表必须对应，否则无法查出数据

使用示例

Hive作业流程示例如下：

• 示例1

a. 编写如下脚本，保存为hiveSample1.sql文件。

CREATE EXTERNAL TABLE kmrusers (
 userid INT,
 movieid INT,
 rating INT,
 unixtime STRING )
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION 'ks3://${Bucket}/users';##此处为重点，将数据放在KS3上 
select * from kmrusers;

测试数据资源可下载如下数据集，并将其上传至KS3对应的目录。
资源下载：点击下载

b. 创建并运行作业。

hive -f sqlfilelocation

注：

• ${bucket}是KS3 Bucket，path是Bucket上的路径，需要您填写实际保存Hive脚本的位置。
• 您可以关联一个已有的集群，也可以自动按需创建一个，然后关联上创建的作业。

• 示例2：以HiBench中的scan为例。

a. 编写如下脚本，上传至KS3。

USE DEFAULT;
set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
set mapreduce.job.maps=12;
set mapreduce.job.reduces=6;
set hive.stats.autogather=false;
DROP TABLE uservisits;
CREATE EXTERNAL TABLE uservisits (sourceIP STRING,destURL STRING,visitDate STRING,adRevenue DOUBLE,userAgent STRING,countryCode STRING,languageCode STRING,searchWord STRING,duration INT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS SEQUENCEFILE LOCATION 'ks3://${bucket}/path';

注：其中${bucket}是您的KS3 Bucket，path是Bucket上的路径，需要您填写实际保存uservisits的位置。

可以下载作业需要的资源，并将其上传至您KS3对应的目录。
测试数据 ：点击下载

b. 在KMR中创建Hive作业，并运行。

CREATE EXTERNAL TABLE kmrusers (
 userid INT,
 movieid INT,
 rating INT,
 unixtime STRING )
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION 'ks3://ks3-hdfs-test-2/users';

Hbase 对接KS3

利用KS3-HDFS将Hbase数据存储到KS3，以此摆脱对HDFS本地盘的依赖，降低存储成本。Hbase写模型使用LSMT，可靠写依赖对wal文件append，同时对性能要求高，因此将wal放到HDFS。

开通KS3-HDFS服务，请参见KS3-HDFS服务开通文档。

可能出现的问题

报错信息：

• master.HMaster: hbase:meta,1.1588230740 is NOT online; state={1588230740 state=OPENING,

解决方法：

• 关闭Hbase服务（仅在刚部署时使用该操作，已经作为生产的请慎用）。
• 登录zookeeper，删除zk中Hbase node。
• 重启Hbase即可。

测试

1. hbase pe --rows=5000000 --oneCon=true --presplit=10 --nomapred randomWrite 10

2023-05-16 11:01:33,505 INFO  [TestClient-2] hbase.PerformanceEvaluation: Mean      = 484.29
Min       = 1.00
Max       = 9333906.00
StdDev    = 31403.17
50th      = 2.00
75th      = 2.00
95th      = 7.00
99th      = 11.00
99.9th    = 98.00
99.99th   = 1681804.71
99.999th  = 3823417.57
2023-05-16 11:01:33,505 INFO  [TestClient-2] hbase.PerformanceEvaluation: No valueSize statistics available
2023-05-16 11:01:33,505 INFO  [TestClient-2] hbase.PerformanceEvaluation: Finished class org.apache.hadoop.hbase.PerformanceEvaluation$RandomWriteTest in 1983275ms at offset 10000000 for 5000000 rows (2.48 MB/s)
2023-05-16 11:01:33,505 INFO  [TestClient-2] hbase.PerformanceEvaluation: Finished TestClient-2 in 1983275ms over 5000000 rows
2023-05-16 11:01:33,505 INFO  [main] hbase.PerformanceEvaluation: [RandomWriteTest] Summary of timings (ms): [1975841, 1962506, 1983275, 1977824, 1983030, 1966597, 1974498, 1965165, 1977491, 1974590]
2023-05-16 11:01:33,594 INFO  [main] hbase.PerformanceEvaluation: [RandomWriteTest duration ]        Min: 1962506ms        Max: 1983275ms        Avg: 1974081ms
2023-05-16 11:01:33,594 INFO  [main] hbase.PerformanceEvaluation: [ Avg latency (us)]        407
2023-05-16 11:01:33,594 INFO  [main] hbase.PerformanceEvaluation: [ Avg TPS/QPS]        25329         row per second

2. hbase pe --rows=500000 --oneCon=true --presplit=10 --nomapred sequentialWrite 10

2023-05-16 14:04:09,300 INFO  [TestClient-9] hbase.PerformanceEvaluation: No valueSize statistics available
2023-05-16 14:04:09,300 INFO  [TestClient-9] hbase.PerformanceEvaluation: Finished class org.apache.hadoop.hbase.PerformanceEvaluation$SequentialWriteTest in 191485ms at offset 4500000 for 500000 rows (2.57 MB/s)
2023-05-16 14:04:09,300 INFO  [TestClient-9] hbase.PerformanceEvaluation: Finished TestClient-9 in 191485ms over 500000 rows
2023-05-16 14:04:09,300 INFO  [main] hbase.PerformanceEvaluation: [SequentialWriteTest] Summary of timings (ms): [151740, 155557, 151781, 158284, 191424, 152046, 191416, 191612, 158239, 191485]
2023-05-16 14:04:09,392 INFO  [main] hbase.PerformanceEvaluation: [SequentialWriteTest duration ]        Min: 151740ms        Max: 191612ms        Avg: 169358ms
2023-05-16 14:04:09,392 INFO  [main] hbase.PerformanceEvaluation: [ Avg latency (us)]        337
2023-05-16 14:04:09,392 INFO  [main] hbase.PerformanceEvaluation: [ Avg TPS/QPS]        29855         row per second
2023-05-16 14:04:09,392 INFO  [main] client.AsyncConnectionImpl: Connection has been closed by main.

3. hbase pe --rows=500000 --oneCon=true --multiGet=10 --nomapred randomRead 20

查询QPS约670

关于KS3-HDFS更多详细信息请参考文档：KS3-HDFS服务概述。