Shulou(Shulou.com)06/02 Report--

This article mainly explains "the method of flink mysql data access". The content of the explanation is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn "the method of flink mysql data access".

I. API access

1. Add dependency

Com.alibaba.ververica

Flink-connector-mysql-cdc

1.1. 0

2.API code

Public static void main (String [] args) throws Exception {

SourceFunction sourceFunction = MySQLSource.builder ()

.hostname ("localhost")

.port (3306)

.databaseList ("test")

.tableList ("test")

.promoializer (new StringDebeziumDeserializationSchema ())

.build ()

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment ()

Env.addSource (sourceFunction)

.print () .setParallelism (1)

Env.execute ()

}

II. SQL access

1. Add jar package to lib

Flink-sql-connector-mysql-cdc_1.1.0.jar

Create tables in 2.mysql

Create table test (

Id INT

Name varchar (100)

Description varchar (100)

Weight DECIMAL (10par 3)

)

3. Create metadata

CREATE TABLE mysql_test (

Id INT NOT NULL

Name STRING

Description STRING

Weight DECIMAL (10par 3)

) WITH (

'connector' = 'mysql-cdc'

'hostname' = 'localhost'

'port' =' 3306'

'username' = 'root'

'password' = 'root'

'database-name' = 'test'

'table-name' = 'test'

);

4. Use query sql

SELECT id, UPPER (name), description, weight FROM mysql_test

5. Add and delete table field tests

Increase does not affect

An error occurs after deleting the table field

[ERROR] Could not execute SQL statement. Reason:

Org.apache.kafka.connect.errors.DataException: name is not a valid field name

Note: if the version of mysql is 8.0, there will be an error in the link on the flink side.

Com.github.shyiko.mysql.binlog.network.AuthenticationException: Client does not support authentication protocol requested by server; consider upgrading MySQL client

The reason for the above problem is that in the previous version of mysql8, the encryption rule was mysql_native_password, while after mysql8, the encryption rule was caching_sha2_password to restore the mysql login password encryption rule to mysql_native_password.

Solution: execute the following command in mysql

Alter user 'root'@'%' identified with mysql_native_password by' root'; modifies authentication rules

Flush privileges; refresh permissions

Thank you for your reading, the above is the content of "the method of flink mysql data access". After the study of this article, I believe you have a deeper understanding of the method of flink mysql data access, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.