Flink写入Kafka

Flink写入Kafka

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.Properties;
public class DataToKafla {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "kafka-test");
        //这里直接把本地文本写入kafka.
        DataStreamSource<String> data = env.readTextFile("datas/1.txt");
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("kafka-test", new SimpleStringSchema(), props);
        data.addSink(producer);
        env.execute();
    }
}