package com.terry.kafkastream;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import java.util.Properties;
/**
* 需求：对数据进行清洗操作
*
* 思路：terry-henshuai 把-清洗掉
*/
public class Application {
public static void a(String[] args) {
//1、定义主题 发送到另外一个主题 数据清洗
String oneTopic = "t1";
String twoTopic = "t2";
//2、设置属性
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"logProcessor");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"bigdata:9092");
//3、实例对象
StreamsConfig streamsConfig = new StreamsConfig(properties);
//4、流计算
Topology topology = new Topology();
//5、定义kafka组件数据源
topology.addSource("Source", oneTopic).addProcessor("Processor", new ProcessorSupplier<byte[], byte[]>() {
@Override
public Processor<byte[], byte[]> get() {
return new LogProcessor();
}
},"Source").addSink("Sink",twoTopic,"Processor");
//6、实例化
KafkaStreams kafkaStreams = new KafkaStreams(topology, properties);
kafkaStreams.start();
}
}
package com.terry.kafkastream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
/**
* 数据清洗
*/
public class LogProcessor implements Processor<byte[], byte[]> {
private ProcessorContext processorContext;
@Override
public void init(ProcessorContext processorContext) {
//传输
this.processorContext=processorContext;
}
@Override
public void process(byte[] key, byte[] value) {
//1、拿到消息数据，专程字符串
String s = new String(value);
//2、如果包含-，则取出
if(s.contains("-")){
String[] split = s.split("-");
s = split[1];
}
processorContext.forward(key,s.getBytes());
}
@Override
public void close() {
}
}
