转载别人的:https://www.jianshu.com/p/f4af986ca164
注：基于别人脚本上，进行针对修改。所以若直接使用清理脚本，请认真看清你的脚本是否是以下形式。如果不是，则脚本需要做适应性调整。
1.编写清理脚本

仅支持这种形式的索引清理
green  open   k8s-stdout-2018.11.01                     PHEIz5NXSw-ljRvOP1cj9Q   5   1    3748246            0      3.2gb          1.6gb
green  open   recom-nginxacclog-2018.10.27              jl5ZBPHsQBifN0-pXEp_yw   5   1   52743481            0     20.8gb         10.4gb
green  open   nginx-json-acclog-2018.11.09              r6jsGcHWRV6RP2mY6zFxDA   5   1      95681            0     50.4mb         25.1mb
green  open   watcher_alarms-2018.11.09                 r_GS_GQGRoCgyOgiCFCuQw   5   1         24            0    323.4kb        161.7kb
green  open   .monitoring-es-6-2018.11.12               o8H2S-iERnuIAWQT7wuBRA   1   1       4842            0      3.2mb          1.6mb
green  open   client-nginxacclog-2018.11.02             FxXdLPpiSnuBtrJOB2BRsw   5   1  179046160            0    220.8gb        110.3gb
green  open   k8s-stderr-2018.11.16                     Vggw7iYCQ6OHtW-sphgGrA   5   1      68546            0     20.4mb         10.2mb
green  open   k8s-stderr-2018.10.21                     ZCeZZFRWSVyyYDKMjzPnbw   5   1      15454            0      5.3mb          2.6mb
green  open   watcher_alarms-2018.10.30                 VqrbMbnuQ3ChPysgnwgm2w   5   1         44            0      371kb        185.5kb

#！/bin/bash
######################################################
# $Name:        clean_es_index.sh
# $Version:     v1.0
# $Function:    clean es log index
# $Author:      sjt
# $Create Date: 2018-05-14
# $Description: shell
######################################################
#本文未加文件锁，需要的可以加
#脚本的日志文件路径
CLEAN_LOG="/root/clean_es_index.log"
#索引前缀
INDEX_PRFIX=$1
DELTIME=$2
if [ "$DELTIME"x = ""x ]; then
   DELTIME=30
fi

echo "ready to delete index $DELTIME ago!!!"

#elasticsearch 的主机ip及端口
SERVER_PORT=10.2.1.2:9200
#取出已有的索引信息
if [ "$INDEX_PRFIX"x = ""x ]; then
     echo  "curl -s \"${SERVER_PORT}/_cat/indices?v\""
     INDEXS=$(curl -s "${SERVER_PORT}/_cat/indices?v" |awk '{print $3}')
else
     echo " curl -s \"${SERVER_PORT}/_cat/indices?v\""
     INDEXS=$(curl -s "${SERVER_PORT}/_cat/indices?v" |grep "${INDEX_PRFIX}"|awk '{print $3}')
fi
#删除多少天以前的日志，假设输入10，意味着10天前的日志都将会被删除
# seconds since 1970-01-01 00:00:00 seconds
SECONDS=$(date -d  "$(date  +%F) -${DELTIME} days" +%s)
#判断日志文件是否存在，不存在需要创建。
if [ ! -f  "${CLEAN_LOG}" ]
then
touch "${CLEAN_LOG}"
fi
#删除指定日期索引
echo "----------------------------clean time is $(date +%Y-%m-%d_%H:%M:%S) ------------------------------" >>${CLEAN_LOG}
for del_index in ${INDEXS}
do
    indexDate=$( echo ${del_index} | awk -F '-' '{print $NF}' )
    # echo "${del_index}"
    format_date=$(echo ${indexDate}| sed 's/\.//g')
    #根据索引的名称的长度进行切割，不同长度的索引在这里需要进行对应的修改
    indexSecond=$( date -d ${format_date} +%s )
    #echo "$SECONDS - $indexSecond = $(( $SECONDS - $indexSecond ))"
    if [ "$SECONDS" -gt "$indexSecond" ]; then
        echo "it will delete ${del_index}......."
        echo "${del_index}" >> ${CLEAN_LOG}
        #取出删除索引的返回结果
        delResult=`curl -s  -XDELETE "${SERVER_PORT}/"${del_index}"?pretty" |sed -n '2p'`
        #写入日志
        echo "clean time is $(date)" >>${CLEAN_LOG}
        echo "delResult is ${delResult}" >>${CLEAN_LOG}
    fi
done

2.加入crontab任务

# 进入crontab编辑模式
crontab -e 
# 在crontab编辑模式中输入
10 1 * * * sh /app/elk/es/es-index-clear.sh > /dev/null 2>&1

log.retention.hours=48 #数据最多保存48小时，默认是7天
log.retention.bytes=1073741824 #数据最多1G，默认是不限制大小。

在logstash.conf配置中，可以使用ruby动态修改某个字段数据。

filter {
    if [type] == "deployment" {
        drop {}
    }

    mutate {
        remove_field => ["kafka"]
    }

    ruby {
        code => "
        timestamp = event.get('@timestamp') #从字段中获取@timestamp字段，
        localtime = timestamp.time + 28800 #加上8个小时偏差
        localtimeStr = localtime.strftime('%Y.%m.%d')
        event.set('localtime', localtimeStr) #保存最新时间
        "
    }
}

filter {
    grok {
        #match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) (?.*) "(?.*?) (?.*?)\?d=(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)" "(?.*?)'}
        #match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) (?.*) "(?.*?) (?.*?)\?d=(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)".*?'}
        match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) \[(?.*)\] "(?.*?) (?.*?)\?(d=)?(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)".*?'}
    }

    if [tags]{
        drop {}
    }

    if [status] != "200" {
        drop {}
    }

    ruby {
        init => "require 'base64'"
        code => "
        string = event.get('string')
        if string
            begin
                b64 = Base64.decode64(string).force_encoding('utf-8')
                #puts b64, event.get('message')
                event.set('b64_decode', b64)
            rescue ArgumentError
                event.set('b64_decode', '')
            end
        else
            event.set('b64_decode', '')
        end
        "
    }

    if [b64_decode == ""]{
        drop {}
    }

    kv {
        source => "b64_decode"
        field_split => "&?"
        value_split => "="
    }

    if [type] == "template" {
        mutate {
            remove_field => ["@timestamp", "@version", "b64_decode", "message", "string", "body_bytes_sent", "timelocal", "http_user_agent", "http_referer", "status", "protocol", "uri", "ver", "remote_user", "remote_addr", "host", "method", "path"]
        }
    } else {
        date {
            match => ["timelocal", "dd/MMM/yyyy:HH:mm:ss Z"]
            target => "@timestamp"
        }

        mutate {
            remove_field => ["@version", "b64_decode", "message", "string", "body_bytes_sent", "timelocal", "http_user_agent", "http_referer", "status", "protocol", "uri", "ver", "remote_user", "remote_addr", "host", "method", "path"]
        }
    }

}