ApacheFlink学习笔记四

package com.nokia.itms.flink.sql;

import java.util.Properties;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.nokia.itms.flink.FlinkConsumer;
import com.nokia.itms.kafka.KafkaConfig;

public class PrintLowPowerDevicesByOverAggregationTableAPI extends FlinkConsumer {
    private static final Logger logger = LoggerFactory.getLogger(PrintLowPowerDevicesByOverAggregationTableAPI.class);

    /** 
    * @param topic
    * @throws
    */
    public PrintLowPowerDevicesByOverAggregationTableAPI(String topic) {
        super(topic);

    }

    @Override
    public void run() {
        Properties kafkaProps = (Properties) KafkaConfig.getInstance().getKafkaProps().clone();
        kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, this.getClass().getSimpleName());
        logger.info("Event max out of orderness = {} {}",MAX_OUT_OF_ORDERNESS.getSize(),MAX_OUT_OF_ORDERNESS.getUnit());

        StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(execEnv, bsSettings);
        logger.info("before local tz={}",tableEnv.getConfig().getLocalTimeZone().toString());


        String createTable = "\n CREATE TABLE rxpower_detail (\n" +
                "    actualTime BIGINT,\n" +
                "    ponInfo ROW(PONRXPower INT),\n" +
                "    deviceInfo ROW(deviceId STRING), \n" +
                "    event_time AS TO_TIMESTAMP(FROM_UNIXTIME(actualTime/1000)) ,\n" +
                // !!! the event time attribute which produced by
                // below TO_TIMESTAMP_LTZ function can't be used in OVER Aggregation, will throw
                // 'OVER windows' ordering in stream mode must be defined on a time attribute' Exception, it's maybe a bug.
                // !!!
//                "    event_time AS TO_TIMESTAMP_LTZ(actualTime,3) ,\n" +
                "    WATERMARK FOR event_time AS event_time - INTERVAL '" +
                  MAX_OUT_OF_ORDERNESS.getSize() +
                "' " +
                MAX_OUT_OF_ORDERNESS.getUnit() + " , \n" +
                " proc_time as PROCTIME() \n" +
                "    )\n" +
                " with ( \n" +
                "'connector' = 'kafka'" + ",\n" +
                "'topic' = '" + this.topic + "',\n" +
                "'properties.bootstrap.servers' = '" + kafkaProps.getProperty("bootstrap.servers")+ "',\n" +
                //"'properties.group.id' = '" + kafkaProps.getProperty("group.id")+ "',\n" +
                "'scan.startup.mode' = 'latest-offset'" + ",\n" +
                //"'scan.startup.mode' = 'group-offsets'" + ",\n" +
                "'format' = 'json'" + ",\n" +
                "'json.fail-on-missing-field' = 'false'" +",\n" +
                "'json.ignore-parse-errors' = 'true'" +"\n" +
                ")" +"\n" ;

        logger.debug(createTable);
        tableEnv.executeSql(createTable);

        String selectSql = "";

        selectSql = ""
                + "\ncreate temporary view tumble_windowed_result as "
                + "\n select window_start as w_s,window_end as w_e,window_time as w_t,deviceId, count(deviceId) as lc "
                + "\n from TABLE (TUMBLE(TABLE rxpower_detail,DESCRIPTOR(event_time), INTERVAL '5' SECOND))"
                + "\n group by window_start,window_end, window_time,deviceId "
                + "\n"
                ;

        logger.info(selectSql);

        tableEnv.executeSql(selectSql);
        selectSql = "" +
                "\n select w_s, w_e, w_t,deviceId, lc, count(deviceId) over " +
                "\n ( PARTITION BY deviceId " +
                "\n order by w_t  " +
                "\n range between INTERVAL '10' second PRECEDING AND CURRENT ROW )" +
                "\n from tumble_windowed_result " +
                "\n";
        logger.info(selectSql);


        Table table2 = tableEnv.sqlQuery(selectSql);

        DataStream<Row> resultStream = tableEnv.toDataStream(table2);


        resultStream.print();

        try {
            execEnv.execute("table api test");
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

    }

    public static void main(String... args) {
        new PrintLowPowerDevicesByOverAggregationTableAPI("topic.periodic").run();
    }
}

Exception in thread "main" org.apache.flink.table.api.TableException: OVER windows' ordering in stream mode must be defined on a time attribute.
	at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.java:159)
	at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:134)
	at org.apache.flink.table.planner.plan.nodes.exec.ExecEdge.translateToPlan(ExecEdge.java:247)
	at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:104)
	at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:134)
	at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:70)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
	at scala.collection.Iterator.foreach(Iterator.scala:937)
	at scala.collection.Iterator.foreach$(Iterator.scala:937)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
	at scala.collection.IterableLike.foreach(IterableLike.scala:70)
	at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
	at scala.collection.TraversableLike.map(TraversableLike.scala:233)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:69)
	at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:165)
	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toStreamInternal(StreamTableEnvironmentImpl.java:439)
	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toStreamInternal(StreamTableEnvironmentImpl.java:434)
	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.java:358)
	at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.java:331)
	at com.nokia.itms.flink.sql.PrintLowPowerDevicesByOverAggregationTableAPI.run(PrintLowPowerDevicesByOverAggregationTableAPI.java:133)
	at com.nokia.itms.flink.sql.PrintLowPowerDevicesByOverAggregationTableAPI.main(PrintLowPowerDevicesByOverAggregationTableAPI.java:147)

        if (orderKeyType instanceof TimestampType
                && ((TimestampType) orderKeyType).getKind() == TimestampKind.ROWTIME) {
            rowTimeIdx = orderKey;
        } else if (orderKeyType instanceof LocalZonedTimestampType
                && ((LocalZonedTimestampType) orderKeyType).getKind() == TimestampKind.PROCTIME) {
            rowTimeIdx = -1;
        } else {
            throw new TableException(
                    "OVER windows' ordering in stream mode must be defined on a time attribute.");
        }