Flink從Kafka讀取數據存入MySQL

運行環境

    os Version:win10
    Flink Version:1.6.2
    Kafka Version:0.9.0.0
    MySQL Version:5.6.21

數據格式

1.74.103.143 2018-12-20 18:12:00 "GET /class/130.html HTTP/1.1" 404 https://search.yahoo.com/search?p=Flink實戰

zookeeper

啟動zookeeper

命令行輸入zkserver(設置好環境變量)

> zkserver

kafka

啟動kafka

kafka 安裝目錄

.\bin\windows\kafka-server-start.bat .\config\server.properties

創建topic

.\bin\windows\ 目錄下

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test001

打開生產者

kafka-console-producer.bat --broker-list localhost:9092 --topic test

打開消費者

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test001 --from-beginning

mysql 建表

create table log_info(
ID INT NOT NULL AUTO_INCREMENT,
IP VARCHAR(50),
TIME VARCHAR(50),
CourseID VARCHAR(10),
Status_Code VARCHAR(10),
Referer VARCHAR(100),
PRIMARY KEY ( ID )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

查看表數據

select count(*) from log_info;

kafka,zookeeper安裝

flink清洗數據