44 lines
1.7 KiB
Java
44 lines
1.7 KiB
Java
package com.thinker.main;
|
|
|
|
import com.thinker.model.Student;
|
|
import com.alibaba.fastjson.JSON;
|
|
import com.thinker.sql.SinkToMySQL;
|
|
import org.apache.flink.api.common.serialization.SimpleStringSchema;
|
|
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
|
|
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
|
|
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
|
|
|
|
import java.util.Properties;
|
|
|
|
/**
|
|
* @author zeekling [lingzhaohui@zeekling.cn]
|
|
* @version 1.0
|
|
* @apiNote
|
|
* @since 2020-05-05
|
|
*/
|
|
public class SinkToMysql {
|
|
|
|
public static void main(String[] args) throws Exception {
|
|
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
|
|
|
|
Properties props = new Properties();
|
|
props.put("bootstrap.servers", "localhost:9092");
|
|
props.put("zookeeper.connect", "localhost:2181");
|
|
props.put("group.id", "metric-group");
|
|
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
|
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
|
props.put("auto.offset.reset", "latest");
|
|
|
|
SingleOutputStreamOperator<Student> student = env.addSource(new FlinkKafkaConsumer<>(
|
|
"student", //这个 kafka topic 需要和上面的工具类的 topic 一致
|
|
new SimpleStringSchema(),
|
|
props)).setParallelism(1)
|
|
.map(string -> JSON.parseObject(string, Student.class)); //Fastjson 解析字符串成 student 对象
|
|
|
|
student.addSink(new SinkToMySQL()); //数据 sink 到 mysql
|
|
|
|
env.execute("Flink add sink");
|
|
}
|
|
|
|
}
|