The proctime attribute ‘rowtime‘ must not replace an existing field.

男娘i 2023-01-10 01:27 115阅读 0赞

故障代碼如下:

// *************************************************************************
    //     USER DATA TYPES
    // *************************************************************************
    
    /*
     * Simple POJO.
     */
    
    
    import java.sql.Timestamp;
    import org.apache.flink.streaming.api.windowing.time.Time;
    /**
     * Simple POJO.
     */
    public class OrderStream
    {
        public int id;
        public Long user;
        public String product;
        public int amount;
        public Long rowtime;
    
        public OrderStream()
        {
        }
    
        public OrderStream(int id,Long user,String product,int amount,Long rowtime)
        {
            this.id=id;
            this.user = user;
            this.product = product;
            this.amount = amount;
            this.rowtime = rowtime;
        }
    
        @Override
        public String toString()
        {
            return "Order{" +
                    "id="+id+
                    ", user=" + user +
                    ", product='" + product + '\'' +
                    ", amount=" + amount +
                    ", ts=" + rowtime +
                    '}';
        }
    }
    
    
    
    DataStream<OrderStream> orderA = env.fromCollection(Arrays.asList(
                    new OrderStream(1, 1L, "beer", 3, 1505529000L), //2017-09-16 10:30:00
                    new OrderStream(2, 1L, "beer", 3, 1505529000L), //2017-09-16 10:30:00
                    new OrderStream(3, 3L, "rubber", 2, 1505527800L),//2017-09-16 10:10:00
                    new OrderStream(4, 3L, "rubber", 2, 1505527800L),//2017-09-16 10:10:00
                    new OrderStream(5, 1L, "diaper", 4, 1505528400L),//2017-09-16 10:20:00
                    new OrderStream(6, 1L, "diaper", 4, 1505528400L)//2017-09-16 10:20:00
            ));
    
    
    
    
            tEnv.createTemporaryView("Orders", orderA,$("id"),$("user"), $("product"), $("amount"), $("rowtime").proctime());

完整報錯如下:

Exception in thread "main" org.apache.flink.table.api.ValidationException: The proctime attribute 'rowtime' must not replace an existing field.
    	at org.apache.flink.table.typeutils.FieldInfoUtils$ExprToFieldInfo.validateProctimeDoesNotReplaceField(FieldInfoUtils.java:606)
    	at org.apache.flink.table.typeutils.FieldInfoUtils$ExprToFieldInfo.createProctimeFieldInfo(FieldInfoUtils.java:599)
    	at org.apache.flink.table.typeutils.FieldInfoUtils$ExprToFieldInfo.visit(FieldInfoUtils.java:564)
    	at org.apache.flink.table.typeutils.FieldInfoUtils$ExprToFieldInfo.visit(FieldInfoUtils.java:536)
    	at org.apache.flink.table.expressions.ApiExpressionVisitor.visit(ApiExpressionVisitor.java:39)
    	at org.apache.flink.table.expressions.UnresolvedCallExpression.accept(UnresolvedCallExpression.java:132)
    	at org.apache.flink.table.typeutils.FieldInfoUtils.lambda$extractFieldInfosByNameReference$7(FieldInfoUtils.java:431)
    	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    	at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
    	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
    	at org.apache.flink.table.typeutils.FieldInfoUtils.extractFieldInfosByNameReference(FieldInfoUtils.java:432)
    	at org.apache.flink.table.typeutils.FieldInfoUtils.extractFieldInformation(FieldInfoUtils.java:266)
    	at org.apache.flink.table.typeutils.FieldInfoUtils.getFieldsInfo(FieldInfoUtils.java:233)
    	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.lambda$asQueryOperation$0(StreamTableEnvironmentImpl.java:384)
    	at java.util.Optional.map(Optional.java:215)
    	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.asQueryOperation(StreamTableEnvironmentImpl.java:383)
    	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.fromDataStream(StreamTableEnvironmentImpl.java:230)
    	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.createTemporaryView(StreamTableEnvironmentImpl.java:262)
    	at DistinctAggregation4.main(DistinctAggregation4.java:35)

The proctime attribute 'rowtime' must not replace an existing field.

原因與解決方案：

<table> 
 <tbody> 
  <tr> 
   <td style="width:290px;">原因</td> 
   <td>解決方案</td> 
  </tr> 
  <tr> 
   <td style="width:290px;">使用proctime，那麼列名不能是OrderStream中的</td> 
   <td>tEnv.createTemporaryView("Orders", orderA,$("id"),$("user"), $("product"), $("amount"), $("rowtime1").proctime());</td> 
  </tr> 
  <tr> 
   <td style="width:290px;">使用rowtime,，列名可以是OrderStream中的．</td> 
   <td> <pre>tEnv.createTemporaryView("Orders", orderA,$("id"),$("user"), $("product"), $("amount"), $("rowtime").rowtime());</pre> </td> 
  </tr> 
 </tbody> 
</table>