Flink Kafka使用者在使用数据流 key时引发空指针异常


0

我使用这个例子Flink CEP,在这里我分离出数据,因为我已经创建了一个应用程序,它正在向Kafka发送应用程序,另一个应用程序正在从Kafka读取数据。。。我生成了类温度警告的生产者,也就是说,在 Kafka ,我正在发送与温度警告相关的数据,下面是我的代码,它正在消耗来自 Kafka 的数据。。。

StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(5000);
Properties properties=new Properties();
properties.setProperty("bootstrap.servers", "PUBLICDNS:9092");
properties.setProperty("zookeeper.connect", "PUBLICDNS:2181");
properties.setProperty("group.id", "test");



DataStream<TemperatureWarning> dstream=env.addSource(new FlinkKafkaConsumer09<TemperatureWarning>("MonitoringEvent", new MonitoringEventSchema(), properties));

Pattern<TemperatureWarning, ?> alertPattern = Pattern.<TemperatureWarning>begin("first")
.next("second")
.within(Time.seconds(20));

PatternStream<TemperatureWarning> alertPatternStream = CEP.pattern(
dstream.keyBy("rackID"),
alertPattern);

DataStream<TemperatureAlert> alerts = alertPatternStream.flatSelect(
(Map<String, TemperatureWarning> pattern, Collector<TemperatureAlert> out) -> {
TemperatureWarning first = pattern.get("first");
TemperatureWarning second = pattern.get("second");

      if (first.getAverageTemperature() &lt; second.getAverageTemperature()) {
          out.collect(new TemperatureAlert(second.getRackID(),second.getAverageTemperature(),second.getTimeStamp()));
      }
  });

dstream.print();
alerts.print();

env.execute("Flink Kafka Consumer");

但当我执行此应用程序时,它会引发以下异常:

Exception in thread "main" java.lang.NullPointerException
at org.apache.flink.api.common.operators.Keys$ExpressionKeys.<init>(Keys.java:329)
at org.apache.flink.streaming.api.datastream.DataStream.keyBy(DataStream.java:274)
at com.yash.consumer.KafkaFlinkConsumer.main(KafkaFlinkConsumer.java:49)

以下是我的课堂温度警告:

public class TemperatureWarning {

private int rackID;
private double averageTemperature;
private long timeStamp;

public TemperatureWarning(int rackID, double averageTemperature,long timeStamp) {
this.rackID = rackID;
this.averageTemperature = averageTemperature;
this.timeStamp=timeStamp;
}

public TemperatureWarning() {
this(-1, -1,-1);
}

public int getRackID() {
return rackID;
}

public void setRackID(int rackID) {
this.rackID = rackID;
}

public double getAverageTemperature() {
return averageTemperature;
}

public void setAverageTemperature(double averageTemperature) {
this.averageTemperature = averageTemperature;
}

public long getTimeStamp() {
return timeStamp;
}

public void setTimeStamp(long timeStamp) {
this.timeStamp = timeStamp;
}

@Override
public boolean equals(Object obj) {
if (obj instanceof TemperatureWarning) {
TemperatureWarning other = (TemperatureWarning) obj;

    return rackID == other.rackID &amp;&amp; averageTemperature == other.averageTemperature;
} else {
    return false;
}

}

@Override
public int hashCode() {
return 41 * rackID + Double.hashCode(averageTemperature);
}

@Override
public String toString() {
//return "TemperatureWarning(" + getRackID() + ", " + averageTemperature + ")";
return "TemperatureWarning(" + getRackID() +","+averageTemperature + ") "+ "," + getTimeStamp();
}

}

以下是我的班级监控计划:

public class MonitoringEventSchema implements DeserializationSchema<TemperatureWarning>,SerializationSchema<TemperatureWarning>

{

@Override
public TypeInformation<TemperatureWarning> getProducedType() {
    // TODO Auto-generated method stub
    return null;
}

@Override
public byte[] serialize(TemperatureWarning element) {
// TODO Auto-generated method stub
return element.toString().getBytes();
}

@Override
public TemperatureWarning deserialize(byte[] message) throws IOException {
// TODO Auto-generated method stub
if(message!=null)
{
String str=new String(message,"UTF-8");
String []val=str.split(",");
TemperatureWarning warning=new TemperatureWarning(Integer.parseInt(val[0]),Double.parseDouble(val[1]),Long.parseLong(val[2]));
return warning;
}
return null;
}

@Override
public boolean isEndOfStream(TemperatureWarning nextElement) {
// TODO Auto-generated method stub
return false;
}

}

现在要做keyBy操作需要什么,就像我提到的流到分区所需的 key一样??这里需要做些什么来解决这个错误??

1 答案


0

问题在于此函数:

@Override
public TypeInformation<TemperatureWarning> getProducedType() {
    // TODO Auto-generated method stub
    return null;
}   

此处不能返回null。


我来回答

写文章

提问题

面试题