public static void main(String[] args){

        System.setProperty("hadoop.home.dir", "D:\\hadoop");

        System.setProperty("es.set.netty.runtime.available.processors", "false");

        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("SendRecord");

        conf.set("spark.streaming.backpressure.enabled", "true");

        conf.set("spark.streaming.receiver.maxRate", "1000");

        conf.set("spark.streaming.kafka.maxRatePerPartition", "1000");

        conf.set("es.nodes", "eshost");

        conf.set("es.port", "9200");

        JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(2));

        Map<String, Object> kafkaParams = new HashMap<>();

        kafkaParams.put("bootstrap.servers", "kafkahost:9092");

        kafkaParams.put("key.deserializer", StringDeserializer.class);

        kafkaParams.put("value.deserializer", StringDeserializer.class);

        kafkaParams.put("group.id", "sparkGroup4");

        kafkaParams.put("auto.offset.reset", "latest");

        kafkaParams.put("enable.auto.commit", false);

        Collection<String> topics = Arrays.asList("users");

        JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream

                (ssc, LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));

        JavaDStream<User> kafkaDStream = stream.map(new Function<ConsumerRecord<String, String>, User>() {

            @Override

            public User call(ConsumerRecord<String, String> record) throws Exception {

                Gson gson = new Gson();

                return gson.fromJson(record.value(), User.class);

            }

        });

        kafkaDStream.foreachRDD(new VoidFunction<JavaRDD<User>>() {

            @Override

            public void call(JavaRDD<User> userJavaRDD) throws Exception {

                userJavaRDD.foreachPartition(new VoidFunction<Iterator<User>>() {

                    @Override

                    public void call(Iterator<User> userIterator) throws Exception {

                        TransportClient client = ESClient.getClient();

                        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();

                        Map<String, Object> map = new HashMap<>();

                        while(userIterator.hasNext()){

                            User user = userIterator.next();

                            map.put("name", user.getName());

                            map.put("age", user.getAge());

                            map.put("desc", user.getDescription());

                            IndexRequest request = client.prepareIndex("users", "info").setSource(map).request();

                            bulkRequestBuilder.add(request);

                        }

                        if(bulkRequestBuilder.numberOfActions() > 0){

                            BulkResponse bulkItemResponses = bulkRequestBuilder.execute().actionGet();

                        }

                    }

                });

            }

        });

        ssc.start();

        try {

            // Wait for the computation to terminate.

            ssc.awaitTermination();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }

public class ESClient {

    public static TransportClient getClient(){

        return Holder.client;

    }

    private static class Holder{

        private static TransportClient client;

        static{

            try {

                Settings setting = Settings.builder()

                        .put("cluster.name", "es")

                        .put("client.transport.sniff", false)

                        .put("client.transport.ping_timeout", "60s")

                        .put("client.transport.nodes_sampler_interval", "60s")

                        .build();

                client = new PreBuiltTransportClient(setting);

                client.addTransportAddress(new TransportAddress(new InetSocketAddress("eshost",9300)));

            } catch (Exception e) {

                System.out.println(e.getMessage());

            }

        }

    }

}