Я новичок в Flink, и сейчас я использую flink 1.7, создавая проект с pyflink для запроса данных из MongoDB и перехода в Mysql. Однако я продолжаю получать ошибку. Причина: java.lang.ClassCastException: невозможно назначить экземпляр org.bson.BsonDocument полю org.apache.flink.connector.mongo..source.MongoSource.filter типа org.bson.BsonDocument в экземпляр org.apache.flink.connector.mongodb.source.MongoSource. И, я предполагаю, что ошибка из-за модуля flink-connector-mongodb, но я с Java не знаком. Кто-нибудь может помочь мне решить эту проблему? Или какие файлы jar я пропустил?
Ниже приведена часть моего кода, включающая использованные мной файлы jar.
env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars(
"file:///opt/flink/lib/flink-sql-connector-mongodb-1.1.0-1.17.jar",
"file:///opt/flink/lib/bson-5.1.2.jar",
"file:///opt/flink/lib/mongodb-driver-sync-5.1.2.jar",
"file:///opt/flink/lib/mongodb-driver-core-5.1.2.jar",
"file:///opt/flink/lib/flink-connector-jdbc-3.1.2-1.17.jar",
"file:///opt/flink/lib/mysql-connector-java-8.0.26.jar",
"file:///opt/flink/lib/slf4j-api-2.0.13.jar",
"file:///opt/flink/lib/slf4j-simple-2.0.13.jar",
)
env.add_classpaths(
"file:///opt/flink/lib/bson-5.1.2.jar",
"file:///opt/flink/lib/mongodb-driver-sync-5.1.2.jar",
"file:///opt/flink/lib/mongodb-driver-core-5.1.2.jar",
)
t_env = StreamTableEnvironment.create(env)
t_env.execute_sql(
"""
CREATE TABLE source (
...
) with (
'connector' = 'mongodb',
...
)
)
t_env.execute_sql(
"""
CREATE TABLE sink (
...
) with (
'connector' = 'jdbc',
...
)
)
table_result = t_env.sql_query('...')
table_result.execute_insert("sink")
env.execute("MyMongoToMysql")
Подробнее здесь: https://stackoverflow.com/questions/788 ... ink-connec
