This question is cross-posted on Stack Overflow https://stackoverflow.com/questions/66815572/could-not-find-a-suitable-table-factory-for-org-apache-flink-table-factories-ca .
I am running a PyFlink program that reads from Hive `mysource` table, does some processing, then writes to Hive `mysink` table. ``` hive (aiinfra)> describe mysource; OK a bigint b bigint hive (aiinfra)> describe mysink; OK c bigint ``` This is my tree. ``` . ├── deps │ ├── flink-sql-connector-hive-2.2.0_2.11-1.12.0.jar ├── hive.py ``` This is the `hive.py`. ```python import os from pyflink.table import * from pyflink.table.catalog import HiveCatalog settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build() t_env = BatchTableEnvironment.create(environment_settings=settings) t_env.get_config().get_configuration().set_string( "pipeline.jars", f"file://{os.getcwd()}/deps/flink-sql-connector-hive-2.2.0_2.11-1.12.0.jar" ) catalog_name = "myhive" default_database = "aiinfra" hive_conf_dir = "/data/apache/hive/apache-hive-2.1.0-bin/conf" hive_catalog = HiveCatalog(catalog_name, default_database, hive_conf_dir) t_env.register_catalog("myhive", hive_catalog) # set the HiveCatalog as the current catalog of the session t_env.use_catalog("myhive") TRANSFORM_DML = """ INSERT INTO mysink SELECT a + b FROM mysource """ t_env.get_config().set_sql_dialect(SqlDialect.DEFAULT) t_env.execute_sql(TRANSFORM_DML).wait() ``` The above program works fine, until I turn the catalog registration logic into SQL. ```python import os from pyflink.table import * from pyflink.table.catalog import HiveCatalog settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build() t_env = BatchTableEnvironment.create(environment_settings=settings) t_env.get_config().get_configuration().set_string( "pipeline.jars", f"file://{os.getcwd()}/deps/flink-sql-connector-hive-2.2.0_2.11-1.12.0.jar" ) CREATE_CATALOG_DDL = """ CREATE CATALOG myhive WITH ( 'type' = 'hive', 'default-database' = 'aiinfra', 'hive-conf-dir' = '/data/apache/hive/apache-hive-2.1.0-bin/conf' ) """ USE_CATALOG_DDL = """ USE CATALOG myhive """ TRANSFORM_DML = """ INSERT INTO mysink SELECT a + b FROM mysource """ t_env.get_config().set_sql_dialect(SqlDialect.HIVE) t_env.execute_sql(CREATE_CATALOG_DDL) t_env.execute_sql(USE_CATALOG_DDL) t_env.get_config().set_sql_dialect(SqlDialect.DEFAULT) t_env.execute_sql(TRANSFORM_DML).wait() ``` Running the latter version `python hive.py` throws exception: ``` Traceback (most recent call last): File "/data/home/pal-flink/chenyisheng14418/feature-pipelines/pyflink/hive.py", line 42, in <module> t_env.execute_sql(SOURCE_DDL_1) File "/data/apache/flink/flink-1.12.0/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 766, in execute_sql File "/data/apache/flink/flink-1.12.0/opt/python/py4j-0.10.8.1-src.zip/py4j/java_gateway.py", line 1286, in __call__ File "/data/apache/flink/flink-1.12.0/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 147, in deco File "/data/apache/flink/flink-1.12.0/opt/python/py4j-0.10.8.1-src.zip/py4j/protocol.py", line 328, in get_return_value py4j.protocol.Py4JJavaError: An error occurred while calling o10.executeSql. : org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.CatalogFactory' in the classpath. Reason: Required context properties mismatch. The following properties are requested: default-database=aiinfra hive-conf-dir=/data/apache/hive/apache-hive-2.1.0-bin/conf type=hive The following factories have been considered: org.apache.flink.table.catalog.GenericInMemoryCatalogFactory at org.apache.flink.table.factories.TableFactoryService.filterByContext(TableFactoryService.java:322) at org.apache.flink.table.factories.TableFactoryService.filter(TableFactoryService.java:190) at org.apache.flink.table.factories.TableFactoryService.findSingleInternal(TableFactoryService.java:143) at org.apache.flink.table.factories.TableFactoryService.find(TableFactoryService.java:113) at org.apache.flink.table.api.internal.TableEnvironmentImpl.createCatalog(TableEnvironmentImpl.java:1078) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:991) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:665) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748) ``` However, it is clear that HiveCatalogFactory is in the classpath. ``` $ jar tf deps/flink-sql-connector-hive-2.2.0_2.11-1.12.0.jar | grep CatalogFactory org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.class ``` I wonder how to fix it? Thanks. Best, Yik San