基于kcat将TPC-DS数据快速导入kafka集群

在大数据领域，TPC-DS是一个常用的基准测试框架。基准测试结果测量单用户模式下的查询响应时间、多用户模式下的查询吞吐量以及给定硬件、操作系统和数据处理系统配置在受控、复杂、多用户决策支持工作负载下的数据维护性能。

本文以centos 7机器为例，介绍如何使用kcat将TPC-DS生成的测试数据快速导入kafka集群。

TPC-DS下载与测试数据生成

下载地址：https://www.tpc.org/tpc_documents_current_versions/download_programs/tools-download-request5.asp?bm_type=TPC-DS&bm_vers=3.2.0&mode=CURRENT-ONLY

在进入下载地址之后，需要填写一系列相关资料，然后下载地址会发送到你的邮箱，并且这个地址从你邮箱打开之后，仅仅只能下载一次。

如果你觉得麻烦，也可以从如下地址下载：https://xzc.info/files/99a43b58-6a3b-4517-a718-dd1b43abfcab-tpc-ds-tool.zip

接下来便可安装依赖与编译代码。

解压后得到：DSGen-software-code-3.2.0rc1

# 安装依赖
sudo yum install gcc make flex bison byacc git

cd DSGen-software-code-3.2.0rc1/tools
make OS=LINUX

基本使用方法可运行如下命令得到：

./dsdgen --help

dsdgen Population Generator (Version 3.2.0)
Copyright Transaction Processing Performance Council (TPC) 2001 - 2021

USAGE: dsdgen [options]

Note: When defined in a parameter file (using -p), parmeters should
use the form below. Each option can also be set from the command
line, using a form of '-param [optional argument]'
Unique anchored substrings of options are also recognized, and 
case is ignored, so '-sc' is equivalent to '-SCALE'

General Options
===============
ABREVIATION =  <s>       -- build table with abreviation <s>
DIR =  <s>               -- generate tables in directory <s>
HELP =  <n>              -- display this message
PARAMS =  <s>            -- read parameters from file <s>
QUIET =  [Y|N]           -- disable all output to stdout/stderr
SCALE =  <n>             -- volume of data to generate in GB
TABLE =  <s>             -- build only table <s>
UPDATE =  <n>            -- generate update data set <n>
VERBOSE =  [Y|N]         -- enable verbose output
PARALLEL =  <n>          -- build data in <n> separate chunks
CHILD =  <n>             -- generate <n>th chunk of the parallelized data
RELEASE =  [Y|N]         -- display the release information
_FILTER =  [Y|N]         -- output data to stdout
VALIDATE =  [Y|N]        -- produce rows for data validation

Advanced Options
===============
DELIMITER =  <s>         -- use <s> as output field separator
DISTRIBUTIONS =  <s>     -- read distributions from file <s>
FORCE =  [Y|N]           -- over-write data files without prompting
SUFFIX =  <s>            -- use <s> as output file suffix
TERMINATE =  [Y|N]       -- end each record with a field delimiter
VCOUNT =  <n>            -- set number of validation rows to be produced
VSUFFIX =  <s>           -- set file suffix for data validation
RNGSEED =  <n>           -- set RNG seed

下面以生成100G测试数据为例：

mkdir ../data-100G

./dsdgen  -scale 100 -dir ../data/

得到数据：

由于我希望给每个表新增一个主键用于在后续导入hudi表时指定，所以还可以给每个表的每条数据的第一列新增主键字段，以store_sales.dat为例：

awk '$0=NR"|"$0' store_sales.dat > store_sales_3.dat

TPC-DS数据导入kafka集群

kcat是非常轻量级的基于c语言实现的kafka客户端，可实现数据快速导入功能。

首先我们需要下载源码并完成编译：

yum install librdkafka-devel make gcc

git clone https://github.com/edenhill/kcat

cd kcat

./configure --prefix=/usr

make

make install

至此即可使用kcat命令将数据导入kafka集群。

首先使用java客户端新建kafka topic：

bin/kafka-topics.sh --bootstrap-server=host117:19092,host120:19092,host121:19092 \
 --create --partitions 6 --replication-factor 3 --topic store_sales_3_100g

数据导入：

 cat store_sales_3.dat | kcat \
-b host117:19092,host120:19092,host121:19092 \
-t store_sales_2_100g \
-P

查看各个topic情况：

总结

基于kcat我们完成了tpcds数据快速导入kafka集群。基于kcat我们还可以更快速地导入其他文本格式的数据。当然，kcat还有更多其它用法，更多可参考：kcat -help。