Flink : kafka to mysql example

mysql的一些命令

show variables like 'character_set_client';#查詢字符集

show databases;

create database if not exists test;

drop database test;

show tables from test;

use test;

show tables from test;

mysql建表

DROP TABLE IF EXISTS `record`;
CREATE TABLE `record` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `channel` varchar(10) COLLATE utf8mb4_bin DEFAULT NULL,
  `mobile` varchar(11) COLLATE utf8mb4_bin DEFAULT NULL,
  `timestamp` timestamp DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

 

啓動kafka

cd ~/workspace/kafka_2.12-2.1.1
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
bin/kafka-server-start.sh  config/server.properties

 

查看kafka topic列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

 

啓動flink job

KafkaToMysqlTest

發送kafka數據

KafkaSendTest

 

查詢mysql記錄

select * from record

發現有數據