{"job": {"content": {"reader": {"name": "kafka","parameter": {"topic": "kafka_es_client_error","groupid": "es2","bootstrapServers": "k1:9094,k2:9094,k3:9094"},"threads": 6},"writer": {"name": "es","parameter": {"host": ["es1:9300,es2:9300,es3:9300"],"index": "client_error_%s","type": "client_error"}}}}
}

public class EsProducer {private final static Logger LOG = LoggerFactory.getLogger(EsProducer.class);private final KafkaConsumer<String, String> consumer;private ExecutorService executorService;private Configuration conf = null;private static int counter = 0;public EsProducer() {String root = System.getProperty("user.dir") + "/conf/";String path = SystemConfigUtils.getProperty("kafka.x.plugins.exec.path");conf = Configuration.from(new File(root + path));Properties props = new Properties();props.put("bootstrap.servers", conf.getString("job.content.reader.parameter.bootstrapServers"));props.put("group.id", conf.getString("job.content.reader.parameter.groupid"));props.put("enable.auto.commit", "true");props.put("auto.commit.interval.ms", "1000");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");consumer = new KafkaConsumer<String, String>(props);consumer.subscribe(Arrays.asList(conf.getString("job.content.reader.parameter.topic")));}public void execute() {executorService = Executors.newFixedThreadPool(conf.getInt("job.content.reader.threads"));while (true) {ConsumerRecords<String, String> records = consumer.poll(100);if (null != records) {executorService.submit(new KafkaConsumerThread(records, consumer));}}}public void shutdown() {try {if (consumer != null) {consumer.close();}if (executorService != null) {executorService.shutdown();}if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {LOG.error("Shutdown kafka consumer thread timeout.");}} catch (InterruptedException ignored) {Thread.currentThread().interrupt();}}class KafkaConsumerThread implements Runnable {private ConsumerRecords<String, String> records;public KafkaConsumerThread(ConsumerRecords<String, String> records, KafkaConsumer<String, String> consumer) {this.records = records;}@Overridepublic void run() {String index = conf.getString("job.content.writer.parameter.index");String type = conf.getString("job.content.writer.parameter.type");for (TopicPartition partition : records.partitions()) {List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);for (ConsumerRecord<String, String> record : partitionRecords) {JSONObject json = JSON.parseObject(record.value());List<Map<String, Object>> list = new ArrayList<>();Map<String, Object> map = new HashMap<>();index = String.format(index, CalendarUtils.timeSpan2EsDay(json.getLongValue("_tm") * 1000L));if (counter < 10) {LOG.info("Index : " + index);counter++;}for (String key : json.keySet()) {if ("_uid".equals(key)) {map.put("uid", json.get(key));} else {map.put(key, json.get(key));}list.add(map);}EsUtils.write2Es(index, type, list);}}}}}

public class EsUtils {private static TransportClient client = null;static {if (client == null) {client = new PreBuiltTransportClient(Settings.EMPTY);}String root = System.getProperty("user.dir") + "/conf/";String path = SystemConfigUtils.getProperty("kafka.x.plugins.exec.path");Configuration conf = Configuration.from(new File(root + path));List<Object> hosts = conf.getList("job.content.writer.parameter.host");for (Object object : hosts) {try {client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(object.toString().split(":")[0]), Integer.parseInt(object.toString().split(":")[1])));} catch (Exception e) {e.printStackTrace();}}}public static void write2Es(String index, String type, List<Map<String, Object>> dataSets) {BulkRequestBuilder bulkRequest = client.prepareBulk();for (Map<String, Object> dataSet : dataSets) {bulkRequest.add(client.prepareIndex(index, type).setSource(dataSet));}bulkRequest.execute().actionGet();// if (client != null) {// client.close();// }}public static void close() {if (client != null) {client.close();}}
}

#!/bin/sh
# <Usage>: ./delete_es_by_day.sh kafka_error_client logsdate 30 </Usage>echo "<Usage>: ./delete_es_by_day.sh kafka_error_client logsdate 30 </Usage>"

index_name=$1
daycolumn=$2
savedays=$3
format_day=$4if [ ! -n "$savedays" ]; thenecho "Oops. The args is not right,please input again...."exit 1
fiif [ ! -n "$format_day" ]; thenformat_day='%Y%m%d'
fisevendayago=`date -d "-${savedays} day " +${format_day}`curl -XDELETE "es1:9200/${index_name}/_query?pretty" -d "
{"query": {"filtered": {"filter": {"bool": {"must": {"range": {"${daycolumn}": {"from": null,"to": ${sevendayago},"include_lower": true,"include_upper": true}}}}}}}
}"

echo "Finished."

联系方式：
邮箱：smartloli.org@gmail.com
Twitter：https://twitter.com/smartloli
QQ群（Hadoop - 交流社区1）：424769183
温馨提示：请大家加群的时候写上加群理由（姓名＋公司/学校），方便管理员审核，谢谢！

热爱生活，享受编程，与君共勉！

作者：哥不是小萝莉 ［关于我］［犒赏］

出处：http://www.cnblogs.com/smartloli/

转载请注明出处，谢谢合作！