多线程模式下使用Blink TableEnvironment

[jun su]

hi all,

多线程模式下执行sql ， 在非聚合sql时报了如下错误：

Caused by: java.lang.NullPointerException
  at java.util.Objects.requireNonNull(Objects.java:203)
  at
org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:141)
  at
org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:106)


已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE()))
解决


但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？

Caused by: java.lang.NullPointerException
at scala.Predef$.Double2double(Predef.scala:365)
at
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81)
at
org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174)
at
GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown
Source)
at GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown
Source)

--
Best,
Jun Su

[Jeff Zhang]

参考zeppelin的做法，每个线程里都调用这个

https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111


jun su <[hidden email]> 于2020年9月14日周一 上午11:54写道：

> hi all,
>
> 多线程模式下执行sql ， 在非聚合sql时报了如下错误：
>
> Caused by: java.lang.NullPointerException
>   at java.util.Objects.requireNonNull(Objects.java:203)
>   at
>
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:141)
>   at
>
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:106)
>
>
>
> 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE()))
> 解决
>
>
> 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？
>
> Caused by: java.lang.NullPointerException
> at scala.Predef$.Double2double(Predef.scala:365)
> at
>
> org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81)
> at
>
> org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174)
> at
> GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown
> Source)
> at GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown
> Source)
>
> --
> Best,
> Jun Su
>

--
Best Regards

Jeff Zhang

[godfrey he]

TableEnvironment 不是多线程安全的。

btw, 你能描述一下你在多线程情况下怎么使用 TableEnvironment 的吗？

Jeff Zhang <[hidden email]> 于2020年9月14日周一 下午12:10写道：

> 参考zeppelin的做法，每个线程里都调用这个
>
>
> https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111
>
>
> jun su <[hidden email]> 于2020年9月14日周一 上午11:54写道：
>
> > hi all,
> >
> > 多线程模式下执行sql ， 在非聚合sql时报了如下错误：
> >
> > Caused by: java.lang.NullPointerException
> >   at java.util.Objects.requireNonNull(Objects.java:203)
> >   at
> >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:141)
> >   at
> >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:106)
> >
> >
> >
> >
> 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE()))
> > 解决
> >
> >
> > 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？
> >
> > Caused by: java.lang.NullPointerException
> > at scala.Predef$.Double2double(Predef.scala:365)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81)
> > at
> >
> >
> org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174)
> > at
> > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown
> > Source)
> > at
> GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown
> > Source)
> >
> > --
> > Best,
> > Jun Su
> >
>
>
> --
> Best Regards
>
> Jeff Zhang
>

[jun su]

hi godfrey,

我们的用法类似zeppelin, 项目形式类似notebook, 在第一次运行笔记时创建env,
再次运行notebook时会创建新线程来构建job运行, 所以我参考了zepplin的做法暂时fix了这个问题

godfrey he <[hidden email]> 于2020年9月17日周四 下午10:07写道：

> TableEnvironment 不是多线程安全的。
>
> btw, 你能描述一下你在多线程情况下怎么使用 TableEnvironment 的吗？
>
> Jeff Zhang <[hidden email]> 于2020年9月14日周一 下午12:10写道：
>
> > 参考zeppelin的做法，每个线程里都调用这个
> >
> >
> >
> https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111
> >
> >
> > jun su <[hidden email]> 于2020年9月14日周一 上午11:54写道：
> >
> > > hi all,
> > >
> > > 多线程模式下执行sql ， 在非聚合sql时报了如下错误：
> > >
> > > Caused by: java.lang.NullPointerException
> > >   at java.util.Objects.requireNonNull(Objects.java:203)
> > >   at
> > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:141)
> > >   at
> > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:106)
> > >
> > >
> > >
> > >
> >
> 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE()))
> > > 解决
> > >
> > >
> > > 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？
> > >
> > > Caused by: java.lang.NullPointerException
> > > at scala.Predef$.Double2double(Predef.scala:365)
> > > at
> > >
> > >
> >
> org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81)
> > > at
> > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174)
> > > at
> > >
> GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown
> > > Source)
> > > at
> > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown
> > > Source)
> > >
> > > --
> > > Best,
> > > Jun Su
> > >
> >
> >
> > --
> > Best Regards
> >
> > Jeff Zhang
> >
>

--
Best,
Jun Su

[Jeff Zhang]

Hi jun su,

如果是自建平台的话，可以考虑用zeppelin的sdk 来提交作业
https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh





jun su <[hidden email]> 于2020年9月18日周五 上午10:59写道：

> hi godfrey,
>
> 我们的用法类似zeppelin, 项目形式类似notebook, 在第一次运行笔记时创建env,
> 再次运行notebook时会创建新线程来构建job运行, 所以我参考了zepplin的做法暂时fix了这个问题
>
> godfrey he <[hidden email]> 于2020年9月17日周四 下午10:07写道：
>
> > TableEnvironment 不是多线程安全的。
> >
> > btw, 你能描述一下你在多线程情况下怎么使用 TableEnvironment 的吗？
> >
> > Jeff Zhang <[hidden email]> 于2020年9月14日周一 下午12:10写道：
> >
> > > 参考zeppelin的做法，每个线程里都调用这个
> > >
> > >
> > >
> >
> https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111
> > >
> > >
> > > jun su <[hidden email]> 于2020年9月14日周一 上午11:54写道：
> > >
> > > > hi all,
> > > >
> > > > 多线程模式下执行sql ， 在非聚合sql时报了如下错误：
> > > >
> > > > Caused by: java.lang.NullPointerException
> > > >   at java.util.Objects.requireNonNull(Objects.java:203)
> > > >   at
> > > >
> > > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:141)
> > > >   at
> > > >
> > > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:106)
> > > >
> > > >
> > > >
> > > >
> > >
> >
> 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE()))
> > > > 解决
> > > >
> > > >
> > > > 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？
> > > >
> > > > Caused by: java.lang.NullPointerException
> > > > at scala.Predef$.Double2double(Predef.scala:365)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174)
> > > > at
> > > >
> > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown
> > > > Source)
> > > > at
> > > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown
> > > > Source)
> > > >
> > > > --
> > > > Best,
> > > > Jun Su
> > > >
> > >
> > >
> > > --
> > > Best Regards
> > >
> > > Jeff Zhang
> > >
> >
>
>
> --
> Best,
> Jun Su
>

--
Best Regards

Jeff Zhang