步骤1 参考使用Kafka客户端，切换到Kafka客户端安装目录。
1. 以Kafka客户端安装用户，登录Kafka安装客户端的节点。
2. 执行以下命令，切换到客户端安装目录。
cd /opt/client
3. 执行以下命令配置环境变量。
source bigdata_env
4. 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户。如果当前集群 未启用Kerberos认证，则无需执行此命令。
kinit 组件业务用户
步骤2 执行以下命令，创建Kafka的Topic。详细的命令使用可以参考管理Kafka主题。
kafka-topics.sh --topic kafkacktest2 --create --zookeeper ZooKeeper角色实例IP:
2181/kafka --partitions 2 --replication-factor 1 说明
● --topic参数值为要创建的Topic名称，本示例创建的名称为kafkacktest2 。
● --zookeeper：ZooKeeper角色实例所在节点IP地址，填写三个角色实例其中任意一个的IP地 址即可。ZooKeeper角色实例所在节点IP获取参考如下。
– MRS 3.x之前版本，单击集群名称，登录集群详情页面，选择“组件管理 > ZooKeeper
> 实例”。查看ZooKeeper角色实例的IP地址。
– MRS 3.x及后续版本，登录FusionInsight Manager，具体请参见访问FusionInsight Manager（MRS 3.x及之后版本）。然后选择“集群 > 待操作的集群名称 > 服务 >
ZooKeeper > 实例”。查看ZooKeeper角色实例的IP地址。
● --partitions主题分区数和--replication-factor主题备份个数不能大于Kafka角色实例数量。
步骤3 参考从零开始使用ClickHouse登录ClickHouse客户端。
1. 执行以下命令，切换到客户端安装目录。
cd /opt/Bigdata/client 2. 执行以下命令配置环境变量。
source bigdata_env
3. 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户，当前用户需要 具有创建ClickHouse表的权限，具体请参见ClickHouse用户及权限管理章节，为 用户绑定对应角色。如果当前集群未启用Kerberos认证，则无需执行本步骤。
kinit 组件业务用户
例如，kinit clickhouseuser。
4. 执行以下命令连接到要导入数据的ClickHouse实例节点。
clickhouse client --host ClickHouse的实例IP --user 登录名 --password 密码 --port ClickHouse的端口号 --database 数据库名 --multiline
步骤4 参考Kafka引擎表使用语法说明，在ClickHouse中创建Kafka引擎表。例如，如下建表 语句在default数据库下，创建表名为kafka_src_tbl3，Topic名为kafkacktest2、消息格 式为JSONEachRow的Kafka引擎表。
create table kafka_src_tbl3 on cluster default_cluster (id UInt32, age UInt32, msg String)
ENGINE=Kafka() SETTINGS
kafka_broker_list='kafka集群broker实例IP1:9092,kafka集群broker实例IP2:9092,kafka集群broker实例 IP3:9092',
kafka_topic_list='kafkacktest2', kafka_group_name='cg12', kafka_format='JSONEachRow';
步骤5 创建ClickHouse本地复制表。例如，如下创建表名为kafka_dest_tbl3的 ReplicatedMergeTree表。
create table kafka_dest_tbl3 on cluster default_cluster ( id UInt32, age UInt32, msg String )
engine = ReplicatedMergeTree('/clickhouse/tables/{shard}/default/kafka_dest_tbl3', '{replica}') partition by age
order by id;
步骤6 创建MATERIALIZED VIEW，该视图会在后台转换Kafka引擎中的数据并将其放入创建 的ClickHouse表中。
create materialized view consumer3 on cluster default_cluster to kafka_dest_tbl3 as select * from kafka_src_tbl3;
步骤7 再次执行步骤1，进入Kafka客户端安装目录。
步骤8 执行以下命令，在Kafka的Topic中产生消息。例如，如下命令向步骤2中创建的Topic 发送消息。
kafka-console-producer.sh --broker-list kafka集群broker实例IP1:9092,kafka集群 broker实例IP2:9092,kafka集群broker实例IP3:9092 --topic kafkacktest2
>{"id":31, "age":30, "msg":"31 years old"}
>{"id":32, "age":30, "msg":"31 years old"}
>{"id":33, "age":30, "msg":"31 years old"}
>{"id":35, "age":30, "msg":"31 years old"}
步骤9 使用ClickHouse客户端登录步骤3中ClickHouse实例节点，查询ClickHouse表数据。例 如，查询kafka_dest_tbl3本地复制表，Kafka消息中的数据已经同步到该表。
select * from kafka_dest_tbl3;
----结束
4.5.4 使用 ClickHouse 数据迁移工具
ClickHouse数据迁移工具可以将某几个ClickHouseServer实例节点上的一个或多个 MergeTree引擎分区表的部分分区迁移至其他ClickHouseServer节点上相同的表中。在 扩容场景中，可以使用该工具将原节点上的部分数据迁移至新增节点上，从而达到扩 容后的数据均衡。
前提条件
● ClickHouse服务运行正常，Zookeeper服务运行正常，迁入、迁出节点的 ClickHouseServer实例状态正常。
● 请确保迁入节点已有待迁移数据表，且确保该表是MergeTree系列引擎的分区 表。
● 创建迁移任务前请确保所有对待迁移数据表的写入任务已停止，且任务启动后，
只允许对待迁移数据表进行查询操作，禁止对该表进行写入、删除等操作，否则 可能会造成迁移前后数据不一致。
● 迁入节点的ClickHouse数据目录有足够的空间。
操作步骤
步骤1 登录Manager，选择“集群 > 服务 > ClickHouse”， 在ClickHouse服务界面单击“数 据迁移”页签，进入数据迁移界面。
步骤2 单击“创建迁移任务”。
步骤3 在创建迁移任务界面，填写迁移任务的相关参数，具体参考如下表4-3。
表4-3 迁移任务参数说明
参数名 参数取值说明
任务名称 填写具体的任务名称。可由字母、数组及下划线组成，长度为1~50 位，且不能与已有的迁移任务相同。
任务类型 ● 定时任务：选择定时任务时，可以设置“开始时间”参数，设定任 务在当前时间以后的某个时间点执行。
● 即时任务：任务启动后立即开始执行。
开始时间 在“任务类型”参数选择“定时任务”时填写，有效值为当前时间以 后的某个时间（最长为90天以后）。
步骤4 在选择迁移节点界面，填写“迁入节点主机名”、“迁出节点主机名”，单击“下一 步”。
说明
● “迁入节点主机名”与“迁出节点主机名”只能各填写一个主机名，不支持多节点迁移。
具体的参数值可以在ClickHouse服务界面单击“实例”页签，查看当前ClickHouseServer实 例所在“主机名称”列获取。
● “带宽上限”为可选参数，若不填写则为无上限，最大可设置为10000MB/s。
步骤5 在选择迁移数据表界面，单击“数据库”后的 ，选择待迁出节点上存在的数据库，
在“数据表”处选择待迁移的数据表，数据表下拉列表中展示的是所选数据库中的 MergeTree系列引擎的分区表。“节点信息”中展示的为当前迁入节点、迁出节点上 ClickHouse服务数据目录的空间使用情况，单击“下一步”。
步骤6 确认任务信息，确认无误后可以单击“提交”提交任务。
数据迁移工具将根据待迁移数据表的大小自动计算需要迁移的分区，数据迁移量则是 计算出的需要迁移的分区总大小。
步骤7 提交迁移任务成功后，单击操作列的“启动”。如果任务类型是即时任务则开始执行 任务，如果是定时任务则开始倒计时。
步骤8 迁移任务执行过程中，可单击“取消”取消正在执行的迁移任务，若取消任务，则会 回退掉迁入节点上已迁移的数据。
可以单击“更多 > 详情”查看迁移过程中的日志信息。
步骤9 迁移完成后，选择“更多 > 结果”查看迁移结果；选择“更多 > 删除”清理 ZooKeeper以及迁出节点上该迁移任务相关的目录。
----结束