请教group windows的order by问题
Nadie ha tomado este issue todavía.
Evaluación
- Dificultad
- 4/5
- Tiempo estimado
- 3-5 días
- Aptitud para principiantes
- 20/100
- Tipo de issue
- Error
- Claridad
- Necesita aclaración
- Estado de actividad
- Estancado
- Área
- stream-processing
Línea de trabajo
Comienza con la consulta SQL proporcionada y la excepción reportada de StreamExecSort.scala; después, sigue la ruta del planificador a través de StreamExecSort.scala y TableEnvironmentImpl.java. Reproduce la ventana agrupada de un segundo con la entrada de ejemplo de Kafka y determina si ORDER BY ROWTIME es compatible con esta forma de consulta. Se considera terminado cuando se haya establecido y comunicado claramente la sintaxis compatible o la limitación.
Escrito por el modelo de indexación a partir del texto del issue.
Descripción
需要根据1秒的window中的ROWTIME进行排序,但是报如下错误:
Exception in thread "main" org.apache.flink.table.api.TableException: Sort on a non-time-attribute field is not supported.
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSort.translateToPlanInternal(StreamExecSort.scala:118)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSort.translateToPlanInternal(StreamExecSort.scala:59)
at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSort.translateToPlan(StreamExecSort.scala:59)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToTransformation(StreamExecSink.scala:184)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:91)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:48)
at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlan(StreamExecSink.scala:48)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:60)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:59)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:59)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:153)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:682)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.insertIntoInternal(TableEnvironmentImpl.java:355)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.insertInto(TableEnvironmentImpl.java:334)
at com.dtstack.flink.sql.exec.FlinkSQLExec.sqlUpdate(FlinkSQLExec.java:94)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.sqlTranslation(ExecuteProcessHelper.java:235)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.getStreamExecution(ExecuteProcessHelper.java:169)
at com.dtstack.flink.sql.Main.main(Main.java:41)
at com.dtstack.flink.sql.launcher.LauncherMain.main(LauncherMain.java:140)
我的脚本如下:
CREATE TABLE MyTable(
name varchar,
after varchar,
jstime bigint,
WATERMARK FOR jstime AS withOffset(jstime,1000)
)WITH(
type ='kafka10',
bootstrapServers ='pro1:9092',
kafka.auto.offset.reset ='latest',
topic ='test1',
parallelism ='1',
sourcedatatype ='json'
);
CREATE TABLE result_user_info(
name varchar,
after varchar,
jstime bigint,
ROWTIME datetime
)WITH(
type ='console',
parallelism ='1'
);
insert into result_user_info(name,after,jstime,ROWTIME) (select name ,
after ,
jstime,ROWTIME from MyTable
group by name,after,jstime,TUMBLE(ROWTIME, INTERVAL '1' SECOND),ROWTIME order by ROWTIME)
kafka消息:
{"name":"testname","jstime":3119988231,"after":"{"khh":"我是khh0003"}"}
想请教是否是语法写错了,还是有别的排序方法,多谢
- Lenguaje dominante
- Java
- Estrellas
- 2k
- Forks
- 913
- Métricas de merge de PR
- Sin PR fusionados en 30 d
Guía de contribución
No hay ninguna guía de contribución indexada para este repositorio
Primeros pasos
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- Abre un pull request que haga referencia al número del issue.
Más de DTStack/flinkStreamSQL
-
Dificultad 5/5 Más de una semana Aptitud para principiantes 20/100
DTStack/flinkStreamSQL#467 · 1 comentario ·
-
这个如何引入使用flink-connector-hive啊 Abierto
Dificultad 4/5 3-5 días Aptitud para principiantes 25/100
DTStack/flinkStreamSQL#437 · 3 comentarios ·
-
Dificultad 3/5 1-2 días Aptitud para principiantes 38/100
DTStack/flinkStreamSQL#432 · 1 comentario ·
-
新版FlinkStreamSQL已经融合进FlinkX项目 Abierto
Dificultad 5/5 Más de una semana Aptitud para principiantes 15/100
DTStack/flinkStreamSQL#431 ·
-
有无对sql 语法校验的方法 Abierto
Dificultad 5/5 Más de una semana Aptitud para principiantes 15/100
DTStack/flinkStreamSQL#430 · 1 comentario ·
Todos los issues de DTStack/flinkStreamSQL
Issues similares
-
bug
Dificultad 2/5 1-3 horas Aptitud para principiantes 85/100
-
Two open-case totals on one screen: the Programs tile says 15,858 and the nav badge says 15,868 Abiertobug frontend maui-pilot
Dificultad 2/5 1-3 horas Aptitud para principiantes 72/100
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 76/100
objectionary/eo-graphs#74 ·
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 72/100
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 65/100