如何查找标识符“elasticsearch-7”的任何工厂实现“org.apache.flink.table.factories.DynamicTableFactory”；在类路径中

Question

我正在尝试使用Table API和Elasticsearch作为水槽创建一个Pyflink应用程序。

 from pyflink.table import TableEnvironment, EnvironmentSettings

 def log_processing():
    env_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
    t_env = TableEnvironment.create(env_settings)
    t_env.get_config().get_configuration().set_string("pipeline.classpaths", "file:///path_to/flink-sql-connector-kafka_2.12-1.13.1.jar;file:///path_to/flink-sql-connector-elasticsearch7_2.11-1.13.1")

    sink_ddl = """
           CREATE TABLE myUserTable (
  user_id STRING,
  user_name STRING,
  uv BIGINT,
  pv BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-7',
  'hosts' = 'http://localhost:9200',
  'index' = 'users'
)
            """
t_env.execute_sql(sink_ddl)
print(sink_ddl)
sink_table = t_env.sql_query("SELECT * FROM myUserTable")
    


if __name__ == '__main__':
    log_processing()

当我试图运行上述代码时，显示以下错误：

`由：org.apache.flink.table.api.api.validationexception：找不到任何实现org.apache的标识符“ elasticsearch-7”的工厂。 flink.table.factories.dynamictable factory'在类路径中。

可用的工厂标识符是：

• 黑洞
• datagen
• 文件系统
• kafka kafka

print upsert-kafka

at org.apache.flink.table.table.table.factories.factoryutil.discoverfactory（factricutil.java：319） at org.apache.flink.table.factories.factoryutil.enrichnomatchingconnectorError（factricutil.java:463）\

如何摆脱此问题。

Answer 1

您能否仔细检查路径 file:///path_to/flink-sql-connector-elasticsearch7_2.11-1.13.1 是否确实存在。似乎缺少 .jar 后缀。