You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by kcz <57...@qq.com> on 2020/07/17 11:29:39 UTC
flink-1.11 集成hive-1.2.1 DDL问题
idea 本地测试
跟hive有关pom依赖
hive-exec flink-connector-hive_2.11
代码如下:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.enableCheckpointing(60*1000, CheckpointingMode.EXACTLY_ONCE);
// 同一时间只允许进行一个检查点
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setStateBackend(new FsStateBackend(path));
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
String name = "myhive";
String defaultDatabase = "situation";
String hiveConfDir = "/load/data/hive/hive-conf"; // a local path
String version = "1.2.1";
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
tableEnv.registerCatalog("myhive", hive);
// set the HiveCatalog as the current catalog of the session
tableEnv.useCatalog("myhive");
tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS stream_tmp");
tableEnv.executeSql("DROP TABLE IF EXISTS stream_tmp.source_table");
报错如下：
Exception in thread "main" java.lang.IncompatibleClassChangeError: Implementing class
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at org.apache.flink.table.planner.delegation.PlannerBase.<init>(PlannerBase.scala:112)
at org.apache.flink.table.planner.delegation.StreamPlanner.<init>(StreamPlanner.scala:48)
at org.apache.flink.table.planner.delegation.BlinkPlannerFactory.create(BlinkPlannerFactory.java:50)
at org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.create(StreamTableEnvironmentImpl.java:130)
at org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(StreamTableEnvironment.java:111)
at com.hive.HiveTest.main(HiveTest.java:33)
Re: flink-1.11 集成hive-1.2.1 DDL问题
Posted by Kurt Young <yk...@gmail.com>.
1.11 把默认planner换成blink了，需要添加下blink planner的依赖
Best,
Kurt
On Mon, Jul 20, 2020 at 11:39 AM Rui Li <li...@gmail.com> wrote:
> stacktrace上看起来是创建blink planner的时候出错的。检查下依赖的blink planner版本是不是正确？
>
> On Fri, Jul 17, 2020 at 7:29 PM kcz <57...@qq.com> wrote:
>
> > idea 本地测试
> > 跟hive有关pom依赖
> > hive-exec flink-connector-hive_2.11
> > 代码如下:
> > StreamExecutionEnvironment env =
> > StreamExecutionEnvironment.getExecutionEnvironment();
> > env.setParallelism(1);
> > env.enableCheckpointing(60*1000, CheckpointingMode.EXACTLY_ONCE);
> > // 同一时间只允许进行一个检查点
> > env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
> >
> > env.setStateBackend(new FsStateBackend(path));
> >
> > StreamTableEnvironment tableEnv =
> > StreamTableEnvironment.create(env);
> >
> > String name = "myhive";
> > String defaultDatabase = "situation";
> > String hiveConfDir = "/load/data/hive/hive-conf"; // a local
> > path
> > String version = "1.2.1";
> >
> > HiveCatalog hive = new HiveCatalog(name, defaultDatabase,
> > hiveConfDir, version);
> > tableEnv.registerCatalog("myhive", hive);
> >
> > // set the HiveCatalog as the current catalog of the session
> > tableEnv.useCatalog("myhive");
> > tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS stream_tmp");
> > tableEnv.executeSql("DROP TABLE IF EXISTS
> > stream_tmp.source_table");
> >
> >
> > 报错如下：
> >
> > Exception in thread "main" java.lang.IncompatibleClassChangeError:
> > Implementing class
> > at java.lang.ClassLoader.defineClass1(Native Method)
> > at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
> > at
> > java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> > at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
> > at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
> > at java.security.AccessController.doPrivileged(Native Method)
> > at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> > at
> >
> org.apache.flink.table.planner.delegation.PlannerBase.<init>(PlannerBase.scala:112)
> > at
> >
> org.apache.flink.table.planner.delegation.StreamPlanner.<init>(StreamPlanner.scala:48)
> > at
> >
> org.apache.flink.table.planner.delegation.BlinkPlannerFactory.create(BlinkPlannerFactory.java:50)
> > at
> >
> org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.create(StreamTableEnvironmentImpl.java:130)
> > at
> >
> org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(StreamTableEnvironment.java:111)
> > at com.hive.HiveTest.main(HiveTest.java:33)
>
>
>
> --
> Best regards!
> Rui Li
>
Re: flink-1.11 集成hive-1.2.1 DDL问题
Posted by Rui Li <li...@gmail.com>.
stacktrace上看起来是创建blink planner的时候出错的。检查下依赖的blink planner版本是不是正确？
On Fri, Jul 17, 2020 at 7:29 PM kcz <57...@qq.com> wrote:
> idea 本地测试
> 跟hive有关pom依赖
> hive-exec flink-connector-hive_2.11
> 代码如下:
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setParallelism(1);
> env.enableCheckpointing(60*1000, CheckpointingMode.EXACTLY_ONCE);
> // 同一时间只允许进行一个检查点
> env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
>
> env.setStateBackend(new FsStateBackend(path));
>
> StreamTableEnvironment tableEnv =
> StreamTableEnvironment.create(env);
>
> String name = "myhive";
> String defaultDatabase = "situation";
> String hiveConfDir = "/load/data/hive/hive-conf"; // a local
> path
> String version = "1.2.1";
>
> HiveCatalog hive = new HiveCatalog(name, defaultDatabase,
> hiveConfDir, version);
> tableEnv.registerCatalog("myhive", hive);
>
> // set the HiveCatalog as the current catalog of the session
> tableEnv.useCatalog("myhive");
> tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS stream_tmp");
> tableEnv.executeSql("DROP TABLE IF EXISTS
> stream_tmp.source_table");
>
>
> 报错如下：
>
> Exception in thread "main" java.lang.IncompatibleClassChangeError:
> Implementing class
> at java.lang.ClassLoader.defineClass1(Native Method)
> at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
> at
> java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
> at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.<init>(PlannerBase.scala:112)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.<init>(StreamPlanner.scala:48)
> at
> org.apache.flink.table.planner.delegation.BlinkPlannerFactory.create(BlinkPlannerFactory.java:50)
> at
> org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.create(StreamTableEnvironmentImpl.java:130)
> at
> org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(StreamTableEnvironment.java:111)
> at com.hive.HiveTest.main(HiveTest.java:33)
--
Best regards!
Rui Li