已安装Kafka并正常运行未启用KerberosRedHat版本：7.4CM和CDH版本：cdh5.13.3kafka版本：3.0.0(0.11.0)Kudu版本：1.5.0 98.2 操作演示
1.准备测试环境
创建测试topic
kafka-topics --create --zookeeper master.gzyh.com:2181,cdh01.gzyh.com:2181,cdh02.gzyh.com:2181 --replication-factor 3 --partitions 3 --topic kafka2kudu_topic
通过Hue使用Impala创建一个Kudu表：
CREATE TABLE ods_deal_daily_kudu (
id STRING COMPRESSION snappy,
name STRING COMPRESSION snappy,
sex STRING COMPRESSION snappy,
city STRING COMPRESSION snappy,
occupation STRING COMPRESSION snappy,
mobile_phone_num STRING COMPRESSION snappy,
fix_phone_num STRING COMPRESSION snappy,
bank_name STRING COMPRESSION snappy,
address STRING COMPRESSION snappy,
marriage STRING COMPRESSION snappy,
child_num INT COMPRESSION snappy,
PRIMARY KEY (id)
)
PARTITION BY HASH PARTITIONS 16
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='master.gzyh.com'
);
2.生产Kafka消息
创建Maven工程，工程的pom.xml文件：
cdh-project com.cloudera 1.0-SNAPSHOT 4.0.0 kafka-demojar kafka-demo http://maven.apache.org UTF-8 org.apache.kafka kafka-clients0.10.2.0 net.sf.json-lib json-lib2.4 jdk15
编写ReadFileToKafka.java文件：
package com.cloudera.nokerberos;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
public class ReadFileToKafka {
public static String confPath = System.getProperty("user.dir") + File.separator + "conf";
public static void main(String[] args) {
if(args.length < 1) {
System.out.print("缺少输入参数，请指定要处理的text文件");
System.exit(1);
}
String filePath = args[0];
BufferedReader reader = null;
try {
Properties appProperties = new Properties();
appProperties.load(new FileInputStream(new File(confPath + File.separator + "app.properties")));
String brokerlist = String.valueOf(appProperties.get("bootstrap.servers"));
String topic_name = String.valueOf(appProperties.get("topic.name"));
Properties props = getKafkaProps();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerlist);
Producer producer = new KafkaProducer(props);
reader = new BufferedReader(new FileReader(filePath));
String tempString = null;
int line = 1;
// 一次读入一行，直到读入null为文件结束
while ((tempString = reader.readLine()) != null) {
String detailJson = parseJSON(tempString);
ProducerRecord record = new ProducerRecord(topic_name, detailJson);
producer.send(record);
line++;
}
reader.close();
producer.flush();
producer.close();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e1) {
}
}
}
}
private static String parseJSON(String tempString) {
if(tempString != null && tempString.length() > 0) {
Map resultMap = null;
String[] detail = tempString.split(" 01");
resultMap = new HashMap<>();
resultMap.put("id", detail[0]);
resultMap.put("name", detail[1]);
resultMap.put("sex", detail[2]);
resultMap.put("city", detail[3]);
resultMap.put("occupation", detail[4]);
resultMap.put("mobile_phone_num", detail[5]);
resultMap.put("fix_phone_num", detail[6]);
resultMap.put("bank_name", detail[7]);
resultMap.put("address", detail[8]);
resultMap.put("marriage", detail[9]);
resultMap.put("child_num", detail[10]);
return JSONObject.fromObject(resultMap).toString();
}
return null;
}
private static Properties getKafkaProps() {
try{
Properties props = new Properties();
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1000); //批量发送消息
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
return props;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
将编写好的代码使用mvn命令打包
在工程目录使用mvn cleanpackage命令进行编译打包 编写脚本run.sh脚本运行jar包
run.sh脚本：
[root@master kafka-run]# vim run.sh
#!/bin/bash
#########################################
# 创建Topic
# kafka-topics --create --zookeeper master.gzyh.com:2181,cdh01.gzyh.com:2181,cdh02.gzyh.com:2181 --replication-factor 3 --partitions 3 --topic ods_deal_daily_topic
########################################
JAVA_HOME=/usr/java/jdk1.8.0_131
#要读取的文件
read_file=$1
for file in `ls lib/*jar`
do
CLASSPATH=$CLASSPATH:$file
done
export CLASSPATH
${JAVA_HOME}/bin/java -Xms1024m -Xmx2048m com.cloudera.nokerberos.ReadFileToKafka $read_file
conf目录下的配置文件app.properties：
[root@master kafka-run]# vim conf/app.properties bootstrap.servers=cdh01.gzyh.com:9092,cdh02.gzyh.com:9092,cdh03.gzyh.com:9092 topic.name=ods_deal_daily_topic
依赖包可以在命令行使用mvn命令导出：
mvn dependency:copy-dependencies -DoutputDirectory=/tmp/lib
3.创建Pipline
在StreamSets创建一个kafka2kudu的PiplinePipline流程中添加Kafka Consumer作为源并配置Kafka基础信息
配置Kafka相关信息，如Broker、ZK及Topic
配置数据格式化方式，写入Kafka的数据为JSON格式，所以这里选择JSON
添加Kudu模块及配置基本信息
配置Kudu的Master、Table、Operation等
Kudu Masters:可以配置多个，多个地址以“,”分割Table Name：如果使用Impala创建的Kudu表则需要添加impala::前缀Field to ColumnMapping：配置Json中key与Kudu表的column的映射关系，如果字段名称一致则不需要配置。DefaultOpertation：设置操作类型如：insert、upsert、delete
Kudu模块高级配置使用默认配置
4.验证启动kafka2kudu的Pipline
运行run.sh脚本向Kafka发送消息
[root@master kafka-run]# sh run.sh ods_user_600.txt
向Kafka发送消息
查看监控信息
查看Kudu的ods_deal_daily_kudu表内容
入库的数据总条数
可以看到ods_deal_daily_kudu表的总条数与准备的测试数据量一致
大数据视频推荐：
CSDN
大数据语音推荐：
企业级大数据技术应用
大数据机器学习案例之推荐系统
自然语言处理
大数据基础
人工智能：深度学习入门到精通
