我正在使用flink流示例,其中机架可用&我想通过机架id计算温度组的总和。下面是我的代码:
static Properties properties=new Properties();
public static Properties getProperties()
{
properties.setProperty("bootstrap.servers", "54.210.139.57:9092");
properties.setProperty("zookeeper.connect", "54.210.139.57:2181");
//properties.setProperty("deserializer.class", "kafka.serializer.StringEncoder");
//properties.setProperty("group.id", "akshay");
properties.setProperty("auto.offset.reset", "earliest");
return properties;
}
@SuppressWarnings("rawtypes")
public static void main(String[] args) throws Exception
{
StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
Properties props=Program.getProperties();
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
/*DataStream<String> dstream=env.addSource(new FlinkKafkaConsumer09<String>("TemperatureEvent",new SimpleStringSchema(), props));
dstream.filter(dstream -> dstream.)*/
DataStream<TemperatureEvent> dstream=env.addSource(new FlinkKafkaConsumer09<TemperatureEvent>("TemperatureEvent", new TemperatureEventSchema(), props));
DataStream<TemperatureEvent> ds1=dstream.keyBy("rackId").sum(1);
ds1.print();
env.execute("Temperature Consumer");
}
当我尝试执行此代码时,它引发以下异常:程序完成时出现以下异常:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at org.apache.flink.client.program.Client.runBlocking(Client.java:248)
at org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:866)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:333)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1192)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1243)
Caused by: java.lang.IndexOutOfBoundsException: Not 0th field selected for a simple type (non-tuple, non-array).
at org.apache.flink.streaming.util.FieldAccessor.create(FieldAccessor.java:78)
at org.apache.flink.streaming.api.functions.aggregation.SumAggregator.<init>(SumAggregator.java:39)
at org.apache.flink.streaming.api.datastream.KeyedStream.sum(KeyedStream.java:292)
at com.yash.main.Program.main(Program.java:38)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
我使用sum(1),因为我的第0个参数是rackid,第1个参数是pojo temperatureevent中定义的温度,如下所示:
public class TemperatureEvent
{
private int rackId;
private double temperature;
private long timeStamp;
public TemperatureEvent()
{
// TODO Auto-generated constructor stub
}
public TemperatureEvent(int rackId, double temperature, long timeStamp) {
super();
this.rackId = rackId;
this.temperature = temperature;
this.timeStamp = timeStamp;
}
public int getRackId() {
return rackId;
}
public void setRackId(int rackId) {
this.rackId = rackId;
}
public double getTemperature() {
return temperature;
}
public void setTemperature(double temperature) {
this.temperature = temperature;
}
public long getTimeStamp() {
return timeStamp;
}
public void setTimeStamp(long timeStamp) {
this.timeStamp = timeStamp;
}
@Override
public String toString() {
//return String.format("TemperatureEvent [rackId=%s, temperature=%s, timeStamp=%s]",rackId, temperature, timeStamp);
String str=getRackId()+","+temperature+","+getTimeStamp();
return str;
}
这个问题的解决办法是什么??我如何计算温度的总和组由rackid??
1条答案
按热度按时间dtcbnfnu1#
如果类型是元组类型,则只能对这些方法使用基于索引的参数。在你的情况下,它应该与
.sum("temperature")
.