关于大数据:懒癌患者基于-YCSB-构造-hudi-upsert-数据集上篇

实时小白一枚，在线求更加弱小和不便的工具

需要

对 hudi 进行 upsert 压测，简略的链路为 flink 生产 kafka 间接灌入 hudi 表，须要结构 10 亿条数据，可管制 insert 和 update 的配比

YCSB 调研

引言：相熟 hudi 的搭档应该理解到，recordKey 相似于主键是 hudi 的一级公民，通过 recordKey 能够疾速定位到须要 update 的数据文件而不必重写所在分区下的所有数据文件。类比于 key-value 的存储，第一工夫想到了 HBase，就从 HBase 搭档理解到了压力测试工具：YCSB

Yahoo! Cloud Serving Benchmark : The goal of the YCSB project is to develop a framework and common set of workloads for evaluating the performance of different“key-value”and“cloud”serving stores

雅虎提供的客户端测试框架，用于评估不同的 key-value 存储和云服务的性能。依据配置文件，自动化结构数据对 db 进行 insert、update、delete、scan、read 压力测试。而本文重点关注结构数据局部

性能速看

两种模式

• load : 数据初始化，所有的 operator 都是 insert
• run : 客户端运行压测，operator 包含 read、scan、insert、update、delete（良好的设计不便前期扩大 kafka db）

要害配置

• writeallfields : 对于 update 操作对应生成的数据是否蕴含所有字段。依据抉择能够灵便配置 hudi 的 PayLoad
• insertorder : 对生成的 key 是以非 hash 形式生成，对生成 update 的 key 十分重要
• requestdistribution : update 数据的 key 的散布形式，前面应用 sequential，自增 id 的形式
• threadcount : 并发数，启动多少个线程并产生成数据
• fieldcount : 生成的列个数（默认名称 field0, field1…）
• zeropadding : 占位符个数，比方首个 key 是 user1，如果设置个数为 5，那么 key 就是 user00001

案例 case

所有测试的根本配置如下 workload：

workload=site.ycsb.workloads.CoreWorkload
readallfields=true
insertorder=ordered
requestdistribution=sequential
threadcount=1
writeallfields=true
fieldcount=7
zeropadding=10

case 1：load 模式

bin/ycsb.sh load basic \        # basic 是默认 db，将后果打印到 终端，可选 redis、mongodb、hbase 等
-P ./workload \                                    # 指向根本配置文件
-p static_col.dt=20220101 \     # 扩大的字段，临时不必思考
-p static_col.ht=02 \           # 扩大的字段，临时不必思考
-p insertstart=0 \              # key 的起始 id，同下面的 workload 配置，这里优先级最高
-p insertcount=2                # 总共插入两条

# 后果如下，写入两条数据，别离是 INSERT op，key 为 user0000000000 和 user0000000001，从 0 开始自增 到 1，而后是统计数据，吞吐量 /s，耗时等
Loading workload...
Starting test.
***************** properties *****************
"writeallfields"="true"
"fieldcount"="7"
"insertcount"="2"
"threadcount"="1"
"readallfields"="true"
"dotransactions"="false"
"static_col.ht"="02"
"requestdistribution"="sequential"
"workload"="site.ycsb.workloads.CoreWorkload"
"zeropadding"="10"
"insertstart"="0"
"static_col.dt"="20220101"
"insertorder"="ordered"
"db"="site.ycsb.BasicDB"
**********************************************
DBWrapper: report latency for each error is false and specific error codes to track for latency are: []
INSERT usertable user0000000000 [field1=2-(4]m9+n7G{;[s:C-350=F36@/=Om8@)5O1$A38&z&#"4Mg<Ao>+02Rm>'b7Nq;V=85~/.(.Ak>(f'*,524"*&4 z<Sy/Dq'2b> field0=,X%=_m<>&5@1+]=9F)7Qq7'<':49':51b65>.;*85h Es#_q0'd;S7$S)0Fe9+f?3t5[+9&r @i'>n/@q35l6Qy&G/:Jw/N{5Eo? field6=3!>5O-*364,*9R!1T#?6:![o7Vw+E35[w''.;000Go<[?43>?$40Nu7U;$Mc=\k2*f!Ps&W/-N!),=O)98d:X99&x$M/59z#:~> field3=:E? &x6"($',(2p3Uy-""4;r-Bm>Q1<V7)J7!*25Y%(821J+$ ,-%l*;p;Ru Pc3O/23x:2 :Fg+Vs7<4$Wk7F5-8z7Sq;Ky:=$, field2=>*>9:."H1$_y*+|+2>1Ms+Zy/0,5S#$9<$Bo;;$1?~1M'%D5)#f>Ta/$6 [i5=0<)f3]{*< ?M-05(=Dc?]-,Ue+I7!>f?N5)\y6 field5=;X=%/~7-x&^q.880 t))~.V=0Ri23x;Hc'_a8E1<Py0Y1,8&+6$1@+ 3b*Ca*8:<Z%*9:(J7#)z;Jm3<.81(*$(5#l)-x:"&/L3(field4="?n?7d>U!94h0La>64"024-("C3%C%3*>=Ke;;<<9n-Ma2$.(Dw+O-<]!4706?z'J#6K%9@u/2l+-(<Z{7W125|&O3#:d&T)7W}. ]
INSERT usertable user0000000001 [field1=;)|.O9?Xu"8<(^7%'v.8h++d9;f6=z!_e>-841$-7&<-t>*8$'f>F#8\u>Fu3%x2-r#Tk3F/59p"&29B9 N}5_w'Fo>B5<Pu7 f* field0=.X34#b&-`'=d8[g*'z#[106&22< 2n'7x3+f>S;0  ,C/)*$7V5(A)&0&9=d(#(-0z#+f+G#.=2$Po9%( >f,Bg/=<-+ 0M/0)$8 field6=6Aw2/`0+p>%.6Bo/I{:1.,6`/N7)92-*61P!3"0-_w$E.Sy0/`68z,:l6\7;,2 J!4@s3=8>Aq>=z:Ni/=f4R+4Gk$Ng"[k;".0 field3=,M% Hw%W6U=/\y'D}&?&$T-6<<9;,=*p3]#.Yo6O'3"&1!&=U9=:l Wg&9`%$|;=4)&b>O&9<50>6G}?I5!$n&=f%Q{*^!$W{> field2=9G-)1p'Sg&; 3!h7Fq4Qw:C1'Rw:;$'Gy64,>)~;Is8<2<&`<\y*;l?H5)R{&)85Uq=Nm>%`=[i'Ey9J{+]=:P#%[)-2~&Lc;Q+0 field5=+#40^q'.z?G;+(|'Gs.Wy/Iq;:"( l75> G97 ,4:h:S55T7#9(,%:$9(5V-))z#J+1"j&9x48~7%f%Wo>@-$Py7%z,58-;$$Xo2 field4=?!r*Hk.Lc'^;&F3-[q$2t#<""..(.<2)$,H?:0<$-")Wu2Q93<t?0f8!b;P/7Xq>T; _=82v'&~?=&0V!1 l) &<G%.Ww.Se3Zm6 ]
[OVERALL], RunTime(ms), 10
[OVERALL], Throughput(ops/sec), 200.0
[TOTAL_GCS_PS_Scavenge], Count, 0
[TOTAL_GC_TIME_PS_Scavenge], Time(ms), 0
[TOTAL_GC_TIME_%_PS_Scavenge], Time(%), 0.0
[TOTAL_GCS_PS_MarkSweep], Count, 0
[TOTAL_GC_TIME_PS_MarkSweep], Time(ms), 0
[TOTAL_GC_TIME_%_PS_MarkSweep], Time(%), 0.0
[TOTAL_GCs], Count, 0
[TOTAL_GC_TIME], Time(ms), 0
[TOTAL_GC_TIME_%], Time(%), 0.0
[CLEANUP], Operations, 1
[CLEANUP], AverageLatency(us), 1.0
[CLEANUP], MinLatency(us), 1
[CLEANUP], MaxLatency(us), 1
[CLEANUP], 95thPercentileLatency(us), 1
[CLEANUP], 99thPercentileLatency(us), 1
[INSERT], Operations, 2
[INSERT], AverageLatency(us), 723.0
[INSERT], MinLatency(us), 186
[INSERT], MaxLatency(us), 1260
[INSERT], 95thPercentileLatency(us), 1260
[INSERT], 99thPercentileLatency(us), 1260
[INSERT], Return=OK, 2

case 2：run 模式

结构 upsert 数据形式，测试的形式是先 load，再 run，比方 case 1 load 了 user0000000000 和 user0000000001 两个 key 的数据，在 run 模式中，反复写入这两条作为更新数据，并且也同时 insert 新 key 的数据从 2 开始

留神：上面的 case 是上自增序列下的时候调研进去的参数含意，并不代表肯定适宜其它的配置

bin/ycsb.sh run basic \                # 应用 run 模式，包含 insert、update、scan 等
-P ./workload \                          # 指向根本配置文件
-p operationcount=4 \                    # insert + update + scan 等总共触发 4 次
-p recordcount=2 \            # insert key 的起始 id 为 2
-p insertstart=0 \            # update key 的起始 id 为 0（为 case 1 曾经初始化的范畴）-p insertcount=2 \            # update key 的最大值不超过 2，即为 0 和 1
-p readproportion=0 \         # 没有读操作
-p updateproportion=0.5 \     # 更新操作占 50%
-p scanproportion=0 \         # 没有扫描操作
-p insertproportion=0.5       # 插入操作占 50%

# 后果如下
# insert 和 update 产生的最终条数是近似 1:1，不是严格的
# update 数据的 key 从 user0000000000 ~ user0000000001
# insert 数据的 key 从 user0000000002 ~ user0000000003
Loading workload...
Starting test.
***************** properties *****************
"insertorder"="ordered"
"updateproportion"="0.5"
"scanproportion"="0"
"writeallfields"="true"
"threadcount"="1"
"operationcount"="2"
"zeropadding"="10"
"readallfields"="true"
"requestdistribution"="sequential"
"dotransactions"="true"
"insertproportion"="0.5"
"insertstart"="0"
"workload"="site.ycsb.workloads.CoreWorkload"
"insertcount"="1"
"readproportion"="0"
"fieldcount"="7"
"recordcount"="1"
"db"="site.ycsb.BasicDB"
**********************************************
DBWrapper: report latency for each error is false and specific error codes to track for latency are: []
UPDATE usertable user0000000000 [field1=(G5)P-.;~#A3%.`"P!/&j7O;?+ )8<42n?Qc3X{2Ws:,:0F3Q5!^a Uk'T9*V}?!v<N{/((7Qw6[1=! ("4<2(,=z;+ <@{-?"! field0=0Us>C{57z2@)=3n"L7*_5(0b-1"!De.-n20t,U%5 : Ig'Sm(Rc+E?:(b2B}%P94V$3z.4j8)v:Ke,%2,Uy9M192d<N>-z:969 field6=*<r*:t03j7O741z?T/-8z';|,>><H!&[o/Bs%^k6 x _k7Jy%#f+@6)b?Ne9Km.Dk-K=1#(,@o"X5/P=(&v+7n/Cw),x"..0Ss+ field3=,0(M#&9.0 r$[-9V32Y7(+<+725)b+)x&_'*#~>N/44`&F)8;2!7l87b:]'8?>%Es43z"$r!'l+>t!7t+<`6_;000=?t+K  &% field2=6T?=_s')b?Ng!4z/V{,$t&H?=?$=S12+$!I19%<97` &8(!|/*j#Ja0O3<O'?B19M197|3#h!_/(Y/<0d?[m32d>V1.U)42t:+,! field5=!((8Ak71>0&h!2j>2p&(|"Tk(G)!Ug'I7<#p4/2<To)@!1Rm1<: 3p"Ks#0n<="5=&3*4!U)'.z7Oa1-p.1f)A50)"!%f"&7T1< field4=),(;<:1l+\9:!"084?@?5>*+Ec#<j+Ye:R'&8>(H{.,0!"~50d#V):Bg'Ua774,&b)C+51.!(|>8b'&`*;b53 %R"V#$Kw';~6]
INSERT usertable user0000000001 [field1=6^e.Q5/<&1-"3Qg/10$Fe3,.)W19Ho'W{&>t#4b3=p'!~?9l/6* >02_s2R#.3b/T16I;%=8:;d3N14:l>Ym3%$(Jo<Wa6N9"4v/ field0=-+d?)l.^i.6|.><5G94<f&L=<341\32X?<?4#("-:0$[q3N;5P}#Py9Ee/Vo7>2(\q6Tw>P?<W#4!,)Ky6Eq)#f'G%4Yi#$d"F7  field6=&F3<Ei'T5%7|-Z75':%7r'Mw">*,&t=-$1S)<Qm.Q#-"61E{9!68I'J#,/v3#6(:b<J99G'2Q/9.~;-b;&p">x#Ia(Eu*9~)8*1 field3==8h*C=>Va2:<1',(Z=2Z99;.5=r**j3Ra"&.'W1)M [a#/4=Z+-Py91x#Yg'8v#S;9S+9Mm3H{3@k9Q?5.,*2*-\9("8$"f,\c? field2=3Gm"7>3;*=Xs>6p*Di':t/Ny.9<0Ca"Sk&%b2O-1Nw<5&0Xq3.6%])7Wy+<6:J{?# 7'&(),?Mi'\g/">8])1<~.Is0Gw73<27n3 field5=$;>=@;71t&]1?Ao#X9C#5)b>@u#]/8920Y/-P=1Xa94:-Lk7A-%Ey3_)$;(. >0$4%!<$9>/I#$Ra:2p;5>6!h#X=>Vg7<t>L9< field4=2*6%,`$^99,.7Q58I98&r"(&4S?52:5%l.^}5F+)"4/R)1X(Xu#Hm>'n*V#5Um67.:Z5):b&%p*3t$S+9/~ 60%.(;<`&Ew(S5$]
[OVERALL], RunTime(ms), 9
[OVERALL], Throughput(ops/sec), 222.22222222222223
[TOTAL_GCS_PS_Scavenge], Count, 0
[TOTAL_GC_TIME_PS_Scavenge], Time(ms), 0
[TOTAL_GC_TIME_%_PS_Scavenge], Time(%), 0.0
[TOTAL_GCS_PS_MarkSweep], Count, 0
[TOTAL_GC_TIME_PS_MarkSweep], Time(ms), 0
[TOTAL_GC_TIME_%_PS_MarkSweep], Time(%), 0.0
[TOTAL_GCs], Count, 0
[TOTAL_GC_TIME], Time(ms), 0
[TOTAL_GC_TIME_%], Time(%), 0.0
[CLEANUP], Operations, 1
[CLEANUP], AverageLatency(us), 1.0
[CLEANUP], MinLatency(us), 1
[CLEANUP], MaxLatency(us), 1
[CLEANUP], 95thPercentileLatency(us), 1
[CLEANUP], 99thPercentileLatency(us), 1
[UPDATE], Operations, 1
[UPDATE], AverageLatency(us), 526.0
[UPDATE], MinLatency(us), 526
[UPDATE], MaxLatency(us), 526
[UPDATE], 95thPercentileLatency(us), 526
[UPDATE], 99thPercentileLatency(us), 526
[UPDATE], Return=OK, 1
[INSERT], Operations, 1
[INSERT], AverageLatency(us), 177.0
[INSERT], MinLatency(us), 177
[INSERT], MaxLatency(us), 177
[INSERT], 95thPercentileLatency(us), 177
[INSERT], 99thPercentileLatency(us), 177
[INSERT], Return=OK, 1

总结

性能上，把 insert 操作作为 insert 数据，update 操作用之前写入过的 key 作为主键，就能够达到须要的成果，并且应用自增 id 的形式，能够不必缓存之前写入过的 key。并且能够灵便的设置 insert 和 update 的配比

然而自增 id 也有毛病，在并发条件下，自增 id 的并发问题就会导致其压测能力不会随着线程数减少而等比回升。所以如果没有这种需要，能够设置 hash key 而不是自增 key

完结但不全完结

懒癌患者比拟难从 0 到 1 写个测试工具，做完以上调研，数据是有了，下篇，将介绍如何基于 ycsb 扩大 db，也不算 db，将魔爪伸向消息中间件 kafka，回到最后的需要（其实也比较简单）

欢送关注公众号：