第三講代碼

導入jar(df_connect_finance-0.1-jar-with-dependencies.jar)包到 /opt/lib/ 下

--------------------------------------------------------->>>>

POST http://localhost:8083/connectors/

body:

--------------------

{

"name": "finace_source_connector_01",

"config": {

"cuid": "finace_source_connector_01",

"connector.class": "com.datafibers.flink.stream.FinanceSourceConnector",

"tasks.max": "1",

"topic": "stock_source",

"portfolio": "Top 10 IT Service",

"interval": "5",

"spoof": "RAND",

"schema.registry.uri": "http://localhost:8081"

}

}

---------------------------

GET http://localhost:8002/subjects

GET http://localhost:8002/subjects/stock_source/versions

GET http://localhost:8002/schemas/ids/1

在第一個窗口輸入：

kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic stock_source

//然后就會不斷有數據更新進來

在第二個窗口打開：mysql -u root -pmypassword

show databases； 查看有沒有simple_stream數據庫

有的話use simple_stream；

沒有的話create simple_stream；

進去看有沒有stock_source

有的話drop掉

然后進入postman：

POST http://localhost:8083/connectors/

body:

---------------

{

"name": "jdbc-sink-mysql-01",

"config": {

"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",

"tasks.max": "1",

"topics": "stock_source",

"connection.url": "jdbc:mysql://localhost:3306/simple_stream",

"connection.user": "root",

"connection.password": "mypassword",

"auto.create": "true"

}

}

-----------------

回到mysql：

show tables；

select count(*) from simple_stream；//查看表內數據量

select * from simple_stream limit 0，10；//查看表內前十條；

總結

以上是生活随笔為你收集整理的kafka 批量 回写mysql_kafka往mysql写数据操作笔记的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。