kafka生产消息，streaming消费

package com.bd.useranalysis.spark.streaming.kafka2es;

import com.alibaba.fastjson.JSON;
import com.bd.useranalysis.common.config.ConfigUtil;
import com.bd.useranalysis.common.project.datatype.DataTypeProperties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.*;

import java.util.*;

public class Kafka2EsJava {

    Properties properties = ConfigUtil.getInstance().getProperties("kafka/kafka-server-config.properties");

    static Set<String> dataTypes = DataTypeProperties.dataTypeMap.keySet();

    public static void main(String[] args) throws InterruptedException {

        SparkConf sparkConf = new SparkConf().setAppName("sparkstreaming_kafka2es").setMaster("local[2]");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        jsc.setLogLevel("WARN");
        JavaStreamingContext jss = new JavaStreamingContext(jsc, Durations.seconds(2L));

        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers","quyf:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "test_20190815");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", true);
        List<String> topicList = Arrays.asList("test","test2");
        JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jss,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topicList, kafkaParams)
        );

        JavaDStream<HashMap<String, String>> recordDS = stream.map(new Function<ConsumerRecord<String, String>, HashMap<String, String>>() {

            @Override
            public HashMap<String, String> call(ConsumerRecord<String, String> record) throws Exception {
                //System.out.println("consumer==>"+record.value());
                return JSON.parseObject(record.value(), HashMap.class);
            }
        });

        for (String type : dataTypes) {
            recordDS.filter(new Function<HashMap<String, String>, Boolean>() {
                @Override
                public Boolean call(HashMap<String, String> resultMap) throws Exception {
                    return resultMap.get("table").equals(type);
                }
            }).foreachRDD(new VoidFunction<JavaRDD<HashMap<String, String>>>() {
                @Override
                public void call(JavaRDD<HashMap<String, String>> mapJavaRDD) throws Exception {
                   mapJavaRDD.foreach(new VoidFunction<HashMap<String, String>>() {
                       @Override
                       public void call(HashMap<String, String> stringStringHashMap) throws Exception {
                           System.out.println(stringStringHashMap.toString());
                       }
                   });
                }
            });
        }

        jss.start();
        jss.awaitTermination();

    }
}

public class GenKafkaData {

    public static void main(String[] args) throws Exception {
        List<String> lines = IOUtils.readLines(new FileReader(
                new File("E:\\wechat\\wechat_source1_1111153.txt")));

        Producer<String, String> producer = getProducer();

        ArrayList<String> columns = DataTypeProperties.dataTypeMap.get("wechat");
        Map<String, String> dataMap = new HashMap<>();
        dataMap.put("table","wechat");
        for(String line : lines){
            String[] fields = line.split("\t");
            for (int i = 0; i < fields.length; i++) {
                dataMap.put(columns.get(i), fields[i]);
            }
            int index = 0;
            while(true){
                String lineRecord = JSON.toJSONString(dataMap);
                producer.send(new ProducerRecord<>("test2",null, lineRecord));
                Thread.sleep(1000);
                index++;
                System.out.println("send->"+lineRecord);
                if(index==10){
                    break;
                }
            }
            //System.out.println("send->"+lineRecord);
            //StringProducer.producer("test", lineRecord);
        }
    }

    public static Producer<String, String> getProducer(){
        Producer<String, String> producer =  new KafkaProducer<String, String>(createProducerProperties());
        return producer;
    }

    private static Properties createProducerProperties() {
        Properties props = new Properties();
//        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("key.serializer", StringSerializer.class);
        props.put("value.serializer", StringSerializer.class);
        props.put("bootstrap.servers", "quyf:9092");
        props.put("linger.ms",1);
        props.put("acks", "all");
        // 消息发送最大尝试次数
        props.put("retries", 0);
        // 一批消息处理大小
        props.put("batch.size", 16384);
        // 增加服务端请求延时
        props.put("linger.ms", 1);
        // 发送缓存区内存大小
        props.put("buffer.memory", 33554432);
        return props;
    }
}