gistfile1.txt

# STAGING
# STAT CUSTOMERS

# CLEANUP PREVIOUS DATA
POST _transform/stat_customers/_stop?force=true
DELETE _transform/stat_customers
DELETE /stat_customers
PUT /_cluster/settings
{"transient": {"logger.org.elasticsearch.xpack.transform.transforms": "debug"}}

---------

# CREATE INDEX
PUT stat_customers
{
    "mappings" : {
        "properties" : {
            "credits" : {"type" : "double"},
            "requests": {"type" : "double"},
            "error_codes.4xx": {"type" : "double"},
            "error_codes.5xx": {"type" : "double"},
            "day": {"type" : "date"},
            "customer": {"type" : "keyword"}
        }
    },
    "settings" : {
        "index" : {
            "number_of_shards" : "1",
            "auto_expand_replicas" : "0-all"
        }
    }
}

# CREATE TRANSFORM
PUT _transform/stat_customers
{
    "description": "Credits per customer",
    "dest": {"index": "stat_customers"},
    "source": {"index": "stats-*"},
    "frequency": "1m",
    "sync": {
        "time": {
            "field": "timestamp",
            "delay": "1m"
        }
    },
    "pivot": {
        "aggs": {
            "credits": { "sum": { "field": "credits" } },
            "requests": { "value_count": { "field": "timestamp" }},
            "error_codes": {
                "scripted_metric": {
                    "init_script": "state.responses = [:]",
                    "map_script": """
                        def key_name = 'missing';
                        if (doc['error.code'].size() != 0) {
                            def error_code = doc['error.code'].value;
                            if (error_code >= 400 && error_code <= 499) {
                                key_name = '4xx';
                            }
                            else if (error_code >= 500 && error_code <= 599) {
                                key_name = '5xx';
                            }
                            else {
                                key_name = error_code;
                            }
                        }

                        if (!state.responses.containsKey(key_name)) {
                            state.responses[key_name] = 0;
                        }
                        state.responses[key_name] += 1;
                        """,
                    "combine_script": "state.responses",
                    "reduce_script": """
                        def counts = [:];
                        for (responses in states) {
                            for (key in responses.keySet()) {
                                if (key == 'missing') {
                                    continue;
                                }
                                if (!counts.containsKey(key)) {
                                    counts[key] = 0;
                                }
                                counts[key] += responses[key];
                            }
                        }
                        return counts;
                        """
                }
            }
        },
        "group_by": {
            "day": {"date_histogram": {"field": "timestamp", "fixed_interval": "1d"}},
            "customer": {"terms": {"field":"customer"}}
        }
    }
}

# START TRANSFORM
POST _transform/stat_customers/_start