hive streaning 问题请教
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.disableOperatorChaining(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); tEnv.getConfig().addConfiguration( new Configuration() .set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(30))); tEnv.executeSql("CREATE TEMPORARY FUNCTION TestFunca AS 'org.example.flink.TestFunca' LANGUAGE JAVA"); tEnv.executeSql("CREATE TABLE datagen (\n" + " name STRING,\n" + " pass STRING,\n" + " type1 INT,\n" + " t1 STRING,\n" + " t2 STRING,\n" + " ts AS localtimestamp,\n" + " WATERMARK FOR ts AS ts\n" + ") WITH (\n" + " 'connector' = 'datagen',\n" + " 'rows-per-second'='1',\n" + " 'fields.type1.min'='1',\n" + " 'fields.type1.max'='10'\n" + ")");
tEnv.getConfig().setSqlDialect(SqlDialect.HIVE); tEnv.executeSql("CREATE TABLE hive_table (\n" + " user_id STRING,\n" + " order_amount STRING\n" + ") PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (\n" + " 'sink.partition-commit.trigger'='partition-time',\n" + " 'sink.partition-commit.delay'='1 h',\n" + " 'sink.partition-commit.policy.kind'='metastore,success-file'\n" + ")");
tEnv.executeSql("insert into hive_table select t1,t2,TestFunca(type1),TestFunca(type1) from datagen");
Caused by: org.apache.flink.table.api.ValidationException: Table options do not contain an option key 'connector' for discovering a connector. at org.apache.flink.table.factories.FactoryUtil.getDynamicTableFactory(FactoryUtil.java:321) at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:157) ... 18 more*来自志愿者整理的flink
写回答
-
You should register a hive catalog first. Hive factory is only work for catalog. You can refer to HiveTableSinkITCase#testStreamingWrite (HiveTableSinkITCase.java). Hope this can help you!*来自志愿者整理的flink
2021-12-06 16:10:20赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
