經過java 向kafka寫入數據，druid從kafka加載數據

一、druid 中配置 加載kafka 任務。能夠參考本人的 相關文章。

二、 向 kafka中發送數據，（注意：在向kafka中發送數據是，必定要注意dataSchema 中timestampSpec的配置，若是配置了時間字段（timestamp）寫入數據時必定要賦值，不然druid將不會加載數據）

package com;

import com.alibaba.fastjson.JSONObject;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 功能描述,該部分必須以中文句號結尾。
 *
 * @author panqingcui
 */
public class VisitorStatisticsProducer {
    private static final Logger LOGGER = LogManager.getLogger();

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.2.176:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 10);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //wikiticker
        //visitor_statistics
        String topic = "visitor_statistics";
        Producer<String, String> producer = new KafkaProducer<>(props);
        List<String> types = new ArrayList();
        types.add("Andriod");
        types.add("Apple Phone");
        types.add("Windows Phone");
       // {"isRobot":false,"channel":"#de.wikipedia","cityName":"Soest","timestamp":"2016-06-27T21:30:50.804Z","flags":"","isUnpatrolled":false,"page":"Fußball-Europameisterschaft 2016/England","countryName":"Germany","regionIsoCode":"NW","diffUrl":"https://de.wikipedia.org/w/index.php?diff=155683760&oldid=155683358","added":93,"metroCode":null,"comment":"alles bei allen Artikeln vereinheitlichen","commentLength":41,"isNew":false,"isMinor":false,"delta":93,"countryIsoCode":"DE","isAnonymous":true,"user":"80.187.112.223","regionName":"North Rhine-Westphalia","deltaBucket":0.0,"deleted":0,"namespace":"Main"}

        for (int i = 0; i < 1; i++) {
            JSONObject jsonObject = new JSONObject();
            String id = UUID.randomUUID().toString();
            jsonObject.put("new_user_id",id);
            if(i%2==0){
                jsonObject.put("user_id", id);
            }else{
                jsonObject.put("click_userid",id);
            }
            if(i%4==0){
                jsonObject.put("is_new",true);
            }else {
                jsonObject.put("is_new",false);
            }
//            jsonObject.put("device_type",types.get(i%types.size()));
//            jsonObject.put("isRobot",false);
//            jsonObject.put("channel","#de.wikipedia");
//            jsonObject.put("cityName","Soest");
            jsonObject.put("timestamp","2017-06-27T21:30:50.804Z");
            Future future = producer.send(new ProducerRecord<String, String>(topic, "test", jsonObject.toJSONString()));
            LOGGER.info(future.get().toString());
        }
        producer.close();
    }
}

三、若是向kafka 發送數據有異常，能夠參見本人java 寫入kafka數據異常。