kafka 批量 回写mysql_kafka往mysql写数据操作笔记
第三講代碼
導(dǎo)入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
在第一個(gè)窗口輸入：
kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic stock_source
//然后就會(huì)不斷有數(shù)據(jù)更新進(jìn)來
在第二個(gè)窗口打開：mysql -u root -pmypassword
show databases； 查看有沒有simple_stream數(shù)據(jù)庫
有的話use simple_stream；
沒有的話create simple_stream；
進(jìn)去看有沒有stock_source
有的話drop掉
然后進(jìn)入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；//查看表內(nèi)數(shù)據(jù)量
select * from simple_stream limit 0，10；//查看表內(nèi)前十條；
總結(jié)
以上是生活随笔為你收集整理的kafka 批量 回写mysql_kafka往mysql写数据操作笔记的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 当兵十二年能拿多少退伍费？
- 下一篇： orm连接mysql_Django项目如