gaogao110 / flink-sql-connector-phoenix Goto Github PK

请问一下，把phoenix版本修改为5.0可以兼容吗？

[root@p65 flink]# bin/sql-client.sh
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/crh/flink-1.13.5-bin-scala_2.11/lib/log4j-slf4j-impl-2.16.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/crh/flink-1.13.5-bin-scala_2.11/lib/phoenix-5.0.0-HBase-2.0-client.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/crh/hadoop-3.2.1/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
2022-06-02 10:08:54,174 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli [] - Found Yarn properties file under /tmp/.yarn-properties-root.
2022-06-02 10:08:54,174 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli [] - Found Yarn properties file under /tmp/.yarn-properties-root.
No default environment specified.
Searching for '/opt/crh/flink-1.13.5-bin-scala_2.11/conf/sql-client-defaults.yaml'...not found.

Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:201)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:161)
Caused by: org.apache.flink.table.api.TableException: Could not instantiate the executor. Make sure a planner module is on the classpath
at org.apache.flink.table.client.gateway.context.ExecutionContext.lookupExecutor(ExecutionContext.java:174)
at org.apache.flink.table.client.gateway.context.ExecutionContext.createTableEnvironment(ExecutionContext.java:115)
at org.apache.flink.table.client.gateway.context.ExecutionContext.(ExecutionContext.java:68)
at org.apache.flink.table.client.gateway.context.SessionContext.create(SessionContext.java:231)
at org.apache.flink.table.client.gateway.local.LocalContextUtils.buildSessionContext(LocalContextUtils.java:100)
at org.apache.flink.table.client.gateway.local.LocalExecutor.openSession(LocalExecutor.java:91)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:88)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:187)
... 1 more
Caused by: org.apache.flink.table.api.TableException: Could not load service provider for table factories.
at org.apache.flink.table.factories.TableFactoryService.discoverFactories(TableFactoryService.java:197)
at org.apache.flink.table.factories.TableFactoryService.findAllInternal(TableFactoryService.java:163)
at org.apache.flink.table.factories.TableFactoryService.findAll(TableFactoryService.java:121)
at org.apache.flink.table.factories.ComponentFactoryService.find(ComponentFactoryService.java:50)
at org.apache.flink.table.client.gateway.context.ExecutionContext.lookupExecutor(ExecutionContext.java:165)
... 8 more
Caused by: java.util.ServiceConfigurationError: org.apache.flink.table.factories.TableFactory: Provider org.apache.flink.connector.org.apache.flink.connector.phoenix.table.PhoenixTableSourceSinkFactory not found
at java.util.ServiceLoader.fail(ServiceLoader.java:239)
at java.util.ServiceLoader.access$300(ServiceLoader.java:185)
at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:372)
at java.util.ServiceLoader$LazyIterator.next(ServiceLoader.java:404)
at java.util.ServiceLoader$1.next(ServiceLoader.java:480)
at java.util.Iterator.forEachRemaining(Iterator.java:116)
at org.apache.flink.table.factories.TableFactoryService.discoverFactories(TableFactoryService.java:193)
... 12 more

Flink SQL> CREATE TABLE pv (

id VARCHAR,
name VARCHAR,
addr VARCHAR,

PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector.type' = 'phoenix',
'connector.url' = 'jdbc:phoenix:node01:2181',
'connector.table' = 'student',
'connector.driver' = 'org.apache.phoenix.jdbc.PhoenixDriver',
'connector.username' = '',
'connector.password' = '',
'phoenix.schema.isnamespacemappingenabled' = 'true',
'phoenix.schema.mapsystemtablestonamespace' = 'true',
'connector.write.flush.max-rows' = '1'
);
[INFO] Execute statement succeed.

Flink SQL> select * from pv;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.client.gateway.SqlExecutionException: Could not execute SQL statement.

你好
Flink SQL> CREATE TABLE test (

sid BIGINT,
ucount BIGINT,
PRIMARY KEY (sid) NOT ENFORCED

) WITH (
'connector.type' = 'phoenix',
'connector.url' = 'jdbc:phoenix:172.16.16.63:2181',
'connector.table' = 'test',
'connector.driver' = 'org.apache.phoenix.jdbc.PhoenixDriver',
'connector.username' = '',
'connector.password' = '',
'phoenix.schema.isnamespacemappingenabled' = 'true',
'phoenix.schema.mapsystemtablestonamespace' = 'true',
'connector.write.flush.max-rows' = '30'
);
[INFO] Execute statement succeed.

Flink SQL> select * from test;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.TableSourceFactory' in
the classpath.

Reason: Required context properties mismatch.

The following properties are requested:
connector.driver=org.apache.phoenix.jdbc.PhoenixDriver
connector.password=
connector.table=test
connector.type=phoenix
connector.url=jdbc:phoenix:172.16.16.63:2181
connector.username=
connector.write.flush.max-rows=30
phoenix.schema.isnamespacemappingenabled=true
phoenix.schema.mapsystemtablestonamespace=true
schema.0.data-type=BIGINT NOT NULL
schema.0.name=sid
schema.1.data-type=BIGINT
schema.1.name=ucount
schema.primary-key.columns=sid
schema.primary-key.name=PK_113901

The following factories have been considered:
org.apache.flink.connector.jdbc.table.JdbcTableSourceSinkFactory
org.apache.flink.connector.hbase2.HBase2TableFactory
org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactory
org.apache.flink.connector.phoenix.table.PhoenixTableSourceSinkFactory
org.apache.flink.table.sources.CsvBatchTableSourceFactory
org.apache.flink.table.sources.CsvAppendTableSourceFactory

可以像红色方框里这样设置kerberos吗

PhoenixDialect类中的这部分代码，其中校验url的开头是不是有误呢，这样jdbc:phoenix开头的url不能校验通过，这种url对于phoenix-core依赖中的驱动似乎不兼容啊

@OverRide
public boolean canHandle(String url) {
return url.startsWith("jdbc:org.apache.flink.connector.phoenix:");
}