ververica / flink-sql-benchmark Goto Github PK

TPCDS_Tools.zip无法下载
具体链接：http://dev.hortonworks.com.s3.amazonaws.com/hive-testbench/tpcds/TPCDS_Tools.zip

hi，最近在写flink 的batch sql，在写的时候遇到了一个问题：
flink 版本：1.10.1
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);
tEnv.registerFunction("explode", new ExplodeFunction());
Table orders = tEnv.from("zhangying480_test");
HiveCatalog catalog = new HiveCatalog("myhive", "test", "", HiveVersionInfo.getVersion());
tEnv.registerCatalog("myhive", catalog);
tEnv.useCatalog("myhive");
Table counts = orders
.groupBy("related_info")
.select("explode(related_info)");
counts.insertInto("zhangying4802_test");
tEnv.execute("test");

如果直接这么执行，会报错如下：
Exception in thread "main" org.apache.flink.table.api.TableException: Only BatchTableSource and InputFormatTableSource are supported in BatchTableEnvironment.
at org.apache.flink.table.plan.nodes.dataset.BatchTableSourceScan.translateToPlan(BatchTableSourceScan.scala:118)
at org.apache.flink.table.plan.nodes.dataset.DataSetDistinct.translateToPlan(DataSetDistinct.scala:84)
at org.apache.flink.table.plan.nodes.dataset.DataSetCalc.translateToPlan(DataSetCalc.scala:92)
at org.apache.flink.table.api.internal.BatchTableEnvImpl.translate(BatchTableEnvImpl.scala:306)
at org.apache.flink.table.api.internal.BatchTableEnvImpl.translate(BatchTableEnvImpl.scala:281)
at org.apache.flink.table.api.internal.BatchTableEnvImpl.writeToSink(BatchTableEnvImpl.scala:145)
at org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:664)
at org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:604)
at org.apache.flink.table.api.internal.TableImpl.insertInto(TableImpl.java:411)
at com.jd.search.algo.invoke.LocalHiveTest.main(LocalHiveTest.java:99)

因此，我用如下代码跑：
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tEnv = TableEnvironment.create(settings);
tEnv.registerFunction("explode", new ExplodeFunction());
Table orders = tEnv.from("zhangying480_test");
HiveCatalog catalog = new HiveCatalog("myhive", "test", "", HiveVersionInfo.getVersion());
tEnv.registerCatalog("myhive", catalog);
tEnv.useCatalog("myhive");
Table counts = orders
.groupBy("related_info")
.select("explode(related_info)");
counts.insertInto("zhangying4802_test");
tEnv.execute("test");
但是这么写的话，tEnv.registerFunction("explode", new ExplodeFunction()); 这行代码报错，查看源码，发现是因为TableEnvironment 只支持ScalarFunction，但是ExplodeFunction继承的是TableFunction。

我们的业务需求是需要使用TableFunction，但是感觉这个问题好像无解了。因为我们使用的是平台的flink 集群，他们的升级比较困难，请问可以在不生版本的同时解决这个问题吗？

I am preparing to test Flink version 1.13.
Can I build based on the dev branch,then modify flink version (1.13-SNAPSHOT) and hive version of pom.xml.
Can it be executed？

when run tpcds_bin_partitioned_orc_1

Caused by: java.lang.IllegalArgumentException: precision 7 is out of range 1 .. 10
at org.apache.orc.TypeDescription.withPrecision(TypeDescription.java:427)
at org.apache.flink.orc.OrcSplitReaderUtil.logicalTypeToOrcType(OrcSplitReaderUtil.java:145)
at org.apache.flink.orc.OrcSplitReaderUtil.logicalTypeToOrcType(OrcSplitReaderUtil.java:177)
at org.apache.flink.orc.OrcSplitReaderUtil.convertToOrcTypeWithPart(OrcSplitReaderUtil.java:118)
at org.apache.flink.orc.OrcSplitReaderUtil.genPartColumnarRowReader(OrcSplitReaderUtil.java:96)
at org.apache.flink.connectors.hive.read.HiveVectorizedOrcSplitReader.(HiveVectorizedOrcSplitReader.java:65)
at org.apache.flink.connectors.hive.read.HiveTableInputFormat.open(HiveTableInputFormat.java:117)
at org.apache.flink.connectors.hive.read.HiveTableInputFormat.open(HiveTableInputFormat.java:56)
at org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:85)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)