????????????????flink1.10.0 ?????????????????????????????????? ???????????? ???????????????????????? tnv.registerDataStream("sqlStream",mysqlinst,'start_ip,'end_ip,'area_id,'unit_id,'device_id) tnv.registerDataStream("OMstream",value,'original_network_id,'asset_id,'types,'d_ip,'d_port,'s_ip,'s_port,'devip,'url,'common_des,'operation_des,'raw_log,'severity,'happen_time,'create_time,'s_ip_num, 'd_ip_num,'method,'asset_area_id,'device_id,'s_mac,'d_mac,'scope,'dcope,'s_asset_id,'d_asset_id,'asset_unit_id,'area_id,'unit_id,'enterprise_id) tnv.registerFunction("ip_to_num",IPtoNum)
?????????? ???????? log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. Exception in thread "main" org.apache.flink.table.api.ValidationException: Too many fields referenced from an atomic type. at org.apache.flink.table.typeutils.FieldInfoUtils.extractFieldInfoFromAtomicType(FieldInfoUtils.java:388) at org.apache.flink.table.typeutils.FieldInfoUtils.extractFieldInformation(FieldInfoUtils.java:259) at org.apache.flink.table.typeutils.FieldInfoUtils.getFieldsInfo(FieldInfoUtils.java:227) at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl$$anonfun$1.apply(StreamTableEnvironmentImpl.scala:237) at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl$$anonfun$1.apply(StreamTableEnvironmentImpl.scala:236) at scala.Option.map(Option.scala:146) at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.asQueryOperation(StreamTableEnvironmentImpl.scala:236) at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.fromDataStream(StreamTableEnvironmentImpl.scala:81) at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.registerDataStream(StreamTableEnvironmentImpl.scala:94) at com.jwell56.networksyslog.jobs.jobsource.MysqlLogToES$.main(MysqlLogToES.scala:77) at com.jwell56.networksyslog.jobs.jobsource.MysqlLogToES.main(MysqlLogToES.scala) ------------------ ???????? ------------------ ??????: "Leonard Xu"<xbjt...@gmail.com>; ????????: 2020??6??10??(??????) ????1:16 ??????: "user-zh"<user-zh@flink.apache.org>; ????: Re: ????flinksql between???? Hi, ????????????????????????source(????)?? ??????????mysql ??????join??????????????????????sql????????????????????????????regular join, ????join??????[1]?? SELECT o.amout, o.currency, r.rate, o.amount * r.rate FROM Orders AS o JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r ON r.currency = o.currency ????JDBC connector????????LookupSource?????????????????????????????????? connector.lookup.cache.ttl ??????????????cache???????????????????????????????????? Best, Leonard Xu [1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/table/sql/queries.html#joins <https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/table/sql/queries.html#joins> [2] https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/table/connect.html#jdbc-connector <https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/table/connect.html#jdbc-connector> > ?? 2020??6??10????10:43???????? <932460...@qq.com> ?????? > > hi,???????? ?????????????????????????? ??????join????????mysql???? ????????????source????????mysql???????????????????????????? > > > > > ------------------&nbsp;????????&nbsp;------------------ > ??????:&nbsp;"wangweigu...@stevegame.cn"<wangweigu...@stevegame.cn&gt;; > ????????:&nbsp;2020??6??9??(??????) ????6:35 > ??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;; > > ????:&nbsp;????: ?????? ????flinksql between???? > > > > > &nbsp; ????1.10???? useBlinkPlanner????????????useOldPlanner???????????? > &nbsp; > ?????????????????? > &nbsp; Exception in thread "main" org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query: > > LogicalProject(num=[$0]) > &nbsp; LogicalJoin(condition=[AND(&gt;($0, $1), <($0, $2))], joinType=[inner]) > &nbsp;&nbsp;&nbsp; FlinkLogicalDataStreamScan(id=[1], fields=[num]) > &nbsp;&nbsp;&nbsp; FlinkLogicalDataStreamScan(id=[2], fields=[startNum, endNum]) > > This exception indicates that the query uses an unsupported SQL feature. > > > > > &nbsp; > ???????? ?????? > ?????????? 2020-06-09 17:41 > ???????? user-zh > ?????? ?????? ????flinksql between???? > hi????????????&amp;nbsp; > 1 flink1.9.0 > 2 oldplanner > <dependency&amp;gt; > <groupId&amp;gt;org.apache.flink</groupId&amp;gt; > <artifactId&amp;gt;flink-table-api-scala_2.11</artifactId&amp;gt; > <version&amp;gt;1.9.0</version&amp;gt; > </dependency&amp;gt; > &nbsp; > &nbsp; > <dependency&amp;gt; > <groupId&amp;gt;org.apache.flink</groupId&amp;gt; > <artifactId&amp;gt;flink-table-planner_2.11</artifactId&amp;gt; > <version&amp;gt;1.9.0</version&amp;gt; > </dependency&amp;gt; > &nbsp; > 3 streaming mode > 4. ???????????? > &amp;nbsp; &amp;nbsp; val sqlStream = env.createInput(jdbcInput) > &amp;nbsp; &amp;nbsp; tnv.registerDataStream("sqlStream",sqlStream,'netstruct_id,'start_ip,'end_ip) > &amp;nbsp; &amp;nbsp; tnv.registerDataStream("OMstream",value,'ip) > //&amp;nbsp; &amp;nbsp; val table = tnv.sqlQuery("select * from&amp;nbsp; OMstream as&amp;nbsp; a left join sqlStream as&amp;nbsp; b on a.ip &amp;gt;b.start_ip and a.ip<b.end_ip") > &amp;nbsp; &amp;nbsp; val table = tnv.sqlQuery("select b.netstruct_id from&amp;nbsp; OMstream as&amp;nbsp; a left join sqlStream as b on a.ip &amp;gt; b.start_ip and a.ip <b.end_ip ") > &amp;nbsp; &amp;nbsp; val resRow = table.toRetractStream[Row] > &nbsp; > 5 ???????????? > Exception in thread "main" org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query:&amp;nbsp; > &nbsp; > &nbsp; > LogicalProject(netstruct_id=[$1]) > &amp;nbsp; LogicalJoin(condition=[AND(&amp;gt;($0, $2), <($0, $3))], joinType=[left]) > &amp;nbsp; &amp;nbsp; FlinkLogicalDataStreamScan(id=[1], fields=[ip]) > &amp;nbsp; &amp;nbsp; FlinkLogicalDataStreamScan(id=[2], fields=[netstruct_id, start_ip, end_ip]) > &nbsp; > &nbsp; > This exception indicates that the query uses an unsupported SQL feature. > Please check the documentation for the set of currently supported SQL features. > at org.apache.flink.table.plan.Optimizer.runVolcanoPlanner(Optimizer.scala:245) > at org.apache.flink.table.plan.Optimizer.optimizeLogicalPlan(Optimizer.scala:160) > at org.apache.flink.table.plan.StreamOptimizer.optimize(StreamOptimizer.scala:66) > at org.apache.flink.table.planner.StreamPlanner.translateToType(StreamPlanner.scala:410) > at org.apache.flink.table.planner.StreamPlanner.org$apache$flink$table$planner$StreamPlanner$$translate(StreamPlanner.scala:182) > at org.apache.flink.table.planner.StreamPlanner$$anonfun$translate$1.apply(StreamPlanner.scala:127) > at org.apache.flink.table.planner.StreamPlanner$$anonfun$translate$1.apply(StreamPlanner.scala:127) > 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.StreamPlanner.translate(StreamPlanner.scala:127) > at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:201) > at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toRetractStream(StreamTableEnvironmentImpl.scala:124) > at org.apache.flink.table.api.scala.TableConversions.toRetractStream(TableConversions.scala:146) > at com.jwell56.linkstarck.LInkStream$.main(LInkStream.scala:37) > at com.jwell56.linkstarck.LInkStream.main(LInkStream.scala) > &nbsp; > &nbsp; > &nbsp; > &nbsp; > &nbsp; > 6 ??????????????&amp;nbsp; > select b.netstruct_id from&amp;nbsp; OMstream as&amp;nbsp; a left join sqlStream as b on a.ip &amp;gt; b.start_ip > ??????????????????????????????&amp;nbsp; > &nbsp; > &nbsp; > ?????? > &nbsp; > &nbsp; > &nbsp; > &nbsp; > ------------------&amp;nbsp;????????&amp;nbsp;------------------ > ??????:&amp;nbsp;"Benchao Li"<libenc...@apache.org&amp;gt;; > ????????:&amp;nbsp;2020??6??9??(??????) ????4:37 > ??????:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;; > &nbsp; > ????:&amp;nbsp;Re: ????flinksql between???? > &nbsp; > &nbsp; > &nbsp; > ???????????????????????? > 1. ????????Flink???????? > 2. ??????planner????blink planner????old planner?? > 3. ??????streaming mode????batch mode?? > 4. ?????????????????????? > &nbsp; > ?????? <932460...@qq.com&amp;gt; ??2020??6??9?????? ????4:26?????? > &nbsp; > &amp;gt; hi??????flinksql?????? select * from a join b on a.ip <b.startip and a.ip > &amp;gt; &amp;amp;gt;b.endip ???????????? ???????????? ??????????????????between??????????????