Построение ElasticSearch и статистика данных

Продукты на платформе имеют анализ данных, который подсчитывает различные анализы данных продавца за определенный период времени (сегодня, вчера, 7 дней, 30 дней...) на платформе.Этот анализ, очевидно, использует счетчик MySql, сумму, Запрос типа GroupBy очень ненадежен, особенно в случае большого количества данных.После обработки MR он сохраняется в HBase.Позже, после общения с друзьями, я был серьезно презираем моими друзьями.Контент презрения в основном в том, что у нас объем данных слишком мал для использования HBase, не говоря уже о MR.Я решил использовать ElasticSearch для реализации ES для краткости.Ну начнем с традиционной конструкции-ямы для добычи-засыпка ям-ямы для добычи-засыпка ям снова.

1. Загрузите, установите и настройте ElasticSearch.

Официальный сайт ElasticSearchwoohoo.elastic.co/products/horrible…Найдите нужную вам версию, я выбрал здесь версию 5.6.9, не спрашивайте почему, ведь в последней версии больше ям в моей неизвестной области! Загрузка 5.6.9 напрямую - самая удобная версия, которую я сейчас использую.ES опирается на минимальную версию JDK1.8, поэтому переменные среды должны быть правильно настроены.

wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.6.9.tar.gz
tar -zxvf elasticsearch-5.6.9.tar.gz -C /usr/local/
cd /usr/local/elasticsearch-5.6.9

Все файлы конфигурации ES находятся в каталоге config, основным файлом конфигурации которого является elasticsearch.yml.После ввода он в основном изменяет несколько мест, а другие места изменяются в соответствии с потребностями бизнеса:

vim config/其中elasticsearch.yml

cluster.name=tsk-es
node.name=tsk1
path.data: /opt/data/elastic/data
path.logs: /opt/data/elastic/log
network.host: 0.0.0.0
http.port: 9200

Не думайте, что после изменения elasticsearch.yml вы сможете напрямую выполнить elasticsearch в каталоге bin, и вам сообщат кучу ошибок! Первая ошибка состоит в том, чтобы сказать вам, что вы не можете запустить ES с пользователем root, поэтому вам нужно сначала создать пользователя.Я создам здесь пользователя с именем elastic и не забудьте дать пользователю разрешение на доступ к папке, а затем su для входа пользователь запускается, но не волнуйтесь. Для эластичных пользователей все еще есть некоторые вещи, которые необходимо изменить:

1. Измените файл /etc/security/limits.conf, в противном случае максимальное количество файловых дескрипторов [4096] для процесса elasticsearch, вероятно, будет слишком низким, будет сообщено об ошибке увеличения как минимум до [65536].

vim /etc/security/limits.conf

* soft nofile 65536
* hard nofile 65536
* soft nproc 2048
* hard nproc 4096

2. Измените файл /etc/sysctl.conf, иначе он сообщит о максимальном количестве областей виртуальной памяти vm.max_map_count [65530], вероятно, слишком мало, увеличьте как минимум до ошибки [262144]

vim /etc/sysctl.conf

vm.max_map_count=262144

После того, как все модификации завершены, вы можете ввести эластичного пользователя для запуска ES.

su elastic
bin/elasticsrarch

Если нет случайностей, ваша ЭП должна работать нормально, тогда используйте браузер для доступа192.168.0.1:9200Вы увидите строку строк JSON, которая доказывает, что ваш ES был успешно запущен.Если вы хотите запустить ES в фоновом режиме, запустите его напрямую

nohup bin/elasticsearch > /opt/data/elastic/elastic.log 2>&1 &

2. Первое понимание работы ЭС

Все операции ES выполняются через HTTP-запросы. Для разных операций используются разные методы и параметры запроса. Например, создание индекса — это PUT, а удаление индекса — DELETE. Если в запросе нет параметров, вы можете напрямую использовать GET. Если есть параметры или использовать POST для отправки данных, то нашим первым шагом должно быть создание индекса для запуска:

PUT:http://192.168.0.1:9200/shopsinfo
{
"mappings":{
    "shopsOrder":{
        "properties":{
            "shopid":{
                "type":"string",
                "index": "not_analyzed"
            },
            "createdate":{
                "type":"string",
                "index": "not_analyzed"
            },
            "timestamp":{
                "type":"long"    
            },
            "paymentType":{
                "type":"string"    ,
                "index": "not_analyzed"
            },
            "amount":{
                "type":"long"
            }
        }
    }
}
}

Вышеупомянутое означает создание индекса с именем магазиныинформация, который содержит сопоставление с именем магазиныОрдер, которое содержит поля shopid, createdata, timestamp, paymentType, сумма и соответствующий тип

Вставка данных относительно проста, просто POST, параметр - JSON.

POST:http://192.168.0.1:9200/shopsinfo/shopsOrder
{
    "shopid": "96119",
    "createdate": "20180410",
    "timestamp": 1523289600000,
    "paymentType": "alipay",
    "amount": 6917
}

удалить данные

POST:http://192.168.0.1:9200/shopsinfo/shopsOrder/_delete_by_query

Запрос
GET/POST:http://192.168.0.1:9200/shopsinfo/shopsOrder/_search

Эксплуатацию ЭС и прочих повторять особо не буду.На официальном сайте есть китайская версия,а на Ду Нианге их много.Самое главное это статистический запрос ЭС,который является ключом к ЭС.

3. Ключевой запрос

ES - это инвертированный индекс, и эффективность запроса очень высока, но оператор запроса ES очень хлопотный.Будь то ES запрос или статистика, он выполняется в виде JSON с BODY POST.Например, я хотите запросить временную метку> Sometime, а shopId равен 100000002 и 100000006 в SQL следующим образом:

select * from shopsOrder where timestamp>1523671189000 and shopid in ("100000002","100000006")

В ES вы должны проверить это так:

POST:http://192.168.0.1:9200/shopsinfo/shopsOrder/_search
{
    "size":20,
    "query":{
        "bool":{
            "must":[
                {
                    "range":{
                        "timestamp":{
                            "gte":1523671189000
                        }

                    }
                },
                {
                    "terms":{
                        "shopid":["100000002","100000006"]
                    }
                }
            ]
        }
    }
}

Я передал здесь параметр size.Если не передать, ES вернет вам 10 фрагментов данных по умолчанию, и ES также вернет вам JSON в результате запроса.В поле hits будет total, то есть общее количество результатов вашего запроса. Вам будут возвращены обращения. содержание результата.

Выше приведен простой запрос.Для статистики ES использует aggs в качестве параметра.Полное название должно называться Aggregation.Например,после запроса только что,я хочу подсчитать общую сумму результата,что похоже на SQL .

select sum(amount)query_amount from shopsOrder where timestamp>1523671189000 and shopid in ("100000002","100000006")

В ES вы должны проверить это так

{
    "aggs":{
        "query_amount":{
            "sum":{
                "field":"amount"
            }
        }
    },
    "query":{
        "bool":{
            "must":[
                {
                    "range":{
                        "timestamp":{
                            "gte":1523671189000
                        }

                    }
                },
                {
                    "terms":{
                        "shopid":["100000002","100000006"]
                    }
                }
            ]
        }
    }
}

Результаты статистики аналогичны этому в query_amount в параметре aggregations возвращаемого значения:

......
"aggregations": {
    "query_amount": {
        "value": 684854
    }
}
......

Далее немного сложнее, и вывод в SQL выглядит следующим образом:

select createdate,sum(amount)query_amount from shopsOrder where timestamp>1523671189000 and shopid in ("100000002","100000006")
group by createdate order by createdate

В ЕС это так:

{
    "size":0,
    "aggs":{
        "orderDate":{
            "terms":{
                "field":"createdate",
                "order":{
                    "_term":"asc"
                }
            },
            "aggs":{
                "query_amount":{
                    "sum":{
                        "field":"amount"
                    }
                }
            }
        }
    },
    "query":{
        "bool":{
            "must":[
                {
                    "range":{
                        "timestamp":{
                            "gte":1523671189000
                        }

                    }
                },
                {
                    "terms":{
                        "shopid":["100000002","100000006"]
                    }
                }
            ]
        }
    }
}

Результат запроса

......
"aggregations": {
    "orderDate": {
        "doc_count_error_upper_bound": 0,
        "sum_other_doc_count": 99,
        "buckets": [
            ......
            {
                "key": "20180415",
                "doc_count": 8,
                "query_amount": {
                    "value": 31632
                }
            },
            {
                "key": "20180417",
                "doc_count": 3,
                "query_amount": {
                    "value": 21401
                }
            },
            {
                "key": "20180418",
                "doc_count": 2,
                "query_amount": {
                    "value": 2333
                }
            }
            ......
        ]
    }
}

Buckets — это результат запроса, key — это значение, сгруппированное по дате создания, doc_count аналогичен count, а query_amount — это значение после суммы. Что касается моего параметра, то есть размер: 0, потому что мне не нужны конкретные записи, которые являются хитами, поэтому я передаю здесь 0

Наконец, давайте перейдем к более сложному: 1. Подсчитайте все итоги 2. Сначала сгруппируйте общую сумму по способу оплаты paymentType, а затем сгруппируйте общую дневную сумму по дням в каждом способе оплаты.

    {
    "size":0,
    "aggs":{
        "amount":{
            "sum":{
                "field":"amount"
            }
        },
        "paymenttype":{
            "terms":{
                "field":"paymentType"
            },
            "aggs":{
                "query_amount":{
                    "sum":{
                        "field":"amount"
                    }
                },
                "payment_date":{
                    "terms":{
                        "field":"createdate"
                    },
                    "aggs":{
                        "query_amount":{
                            "sum":{
                                "field":"amount"
                            }
                        }
                    }
                }
            }
        }
    },
    "query":{
        "bool":{
            "must":[
                {
                    "range":{
                        "timestamp":{
                            "gte":1523671189000
                        }

                    }
                },
                {
                    "terms":{
                        "shopid":["100000002","100000006"]
                    }
                }
            ]
        }
    }
}

Результат запроса:

......
"amount": {
    "value": 684854
},
"paymenttype":{
     ......
    "buckets": [
        {
            "key": "wechatpay",
            "doc_count": 73,
            "amount": {
                "value": 351142
            },
            "payment_date": {
                "doc_count_error_upper_bound": 0,
                "sum_other_doc_count": 25,
                "buckets": [
                    ......
                    {
                        "key": "20180415",
                        "doc_count": 6,
                        "amount": {
                            "value": 29032
                        }
                    },
                    {
                        "key": "20180425",
                        "doc_count": 6,
                        "amount": {
                            "value": 21592
                        }
                    }
                    ......
                ]
            }
        },
        {
            "key": "alipay",
            "doc_count": 67,
            "amount": {
                "value": 333712
            },
            "payment_date": {
                "doc_count_error_upper_bound": 0,
                "sum_other_doc_count": 23,
                "buckets": [
                    ......
                    {
                        "key": "20180506",
                        "doc_count": 8,
                        "amount": {
                            "value": 38280
                        }
                    },
                    {
                        "key": "20180426",
                        "doc_count": 6,
                        "amount": {
                            "value": 41052
                        }
                    }
                    ......
                ]
            }
        }
    ]
}

4, операция JAVA ES

Загрузите соответствующую версию пакета JAR в соответствии с загруженной вами версией ES, я установил 5.6.9, поэтому моя версия пакета JAR также должна быть 5.6.9.

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>5.6.9</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>transport</artifactId>
    <version>5.6.9</version>
</dependency>

Создайте хелпер для работы с ES.Так как мой проект ES основан на SpringBoot, мой хелпер решил управляться с помощью spring.На самом деле, также можно написать синглтон, сначала создайте клиентское соединение

import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import java.net.InetAddress;

Settings settings = Settings.builder().put("cluster.name", "tsk-es").put("client.transport.sniff", true)
                    .build();
TransportClient client = new PreBuiltTransportClient(settings)
                    .addTransportAddresses(new InetSocketTransportAddress(InetAddress.getByName(HOST), PORT));

Вставка данных относительно проста, вы можете напрямую вставить строку JSON или передать JAVA BEAN

import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.common.xcontent.XContentType;

IndexResponse response = client.prepareIndex(index, mapping).setSource(jsonStr, XContentType.JSON)
                .get();

Запрос более сложный.В последнем запросе выше общая сумма группируется по методу оплаты paymentType, а общая сумма каждого дня группируется по дням в каждом способе оплаты.Например:

public void getAmountData(Long startTimestamp, String... shopIds) {
    // 先初始化一个SearchRequestBuilder，指向shopsInfo/shopsOrder
    SearchRequestBuilder sbuilder = client.prepareSearch("shopsinfo").setTypes("shopsOrder");
    // 条件一
    TermsQueryBuilder mpq = QueryBuilders.termsQuery("shopid", shopIds);
    // 条件二
    RangeQueryBuilder mpq2 = QueryBuilders.rangeQuery("timestamp").gte(startTimestamp);
    // 初始化QueryBuilder
    QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(mpq).must(mpq2);
    // 将QueryBuilder放入SearchRequestBuilder
    sbuilder.setQuery(queryBuilder);
    sbuilder.setSize(0);

    // sum，这里创建一个实例全部用这个实例就行
    SumAggregationBuilder salaryAgg = AggregationBuilders.sum("query_amount").field("amount");

    TermsAggregationBuilder paymentAgg = AggregationBuilders.terms("paymentType").field("paymentType");
    paymentAgg.size(100);
    paymentAgg.subAggregation(salaryAgg);
    TermsAggregationBuilder groupDateAff = AggregationBuilders.terms("payment_date").field("createdate")
            .order(Order.term(true));
    groupDateAff.size(100);
    groupDateAff.subAggregation(salaryAgg);
    paymentAgg.subAggregation(groupDateAff);
    // 将统计查询放入SearchRequestBuilder
    sbuilder.addAggregation(salaryAgg).addAggregation(paymentAgg);
    SearchResponse response = sbuilder.execute().actionGet();
    Map<String, Aggregation> aggMap = response.getAggregations().asMap();
    // 获取全部的总额
    InternalSum shopGroupAllAmount = (InternalSum) aggMap.get("amount");
    Double amount = shopGroupAllAmount.getValue();
    ......
}

Всю информацию уже можно получить в SearchResponse, а как парсить данные в дальнейшем, зависит от конкретных потребностей бизнеса и привычек каждого. Различные операции ES просты, сложны и сложны.По словам моих друзей, легко использовать, когда вы знакомы с ним.Это действительно то же самое. Например, когда я делаю статистический запрос, возвращаемые результаты всегда намного меньше ожидаемых результатов, а в поле sum_other_doc_count всегда появляется небольшое число.То же значение по умолчанию 10!

Наконец, я хотел бы поблагодарить своего друга, хотя он и насмехался над тем, что я хотел создавать различные платформы для больших данных (потому что объем наших данных действительно мал), он все же выдал лучшее решение, которое он мог придумать.

Придерживайтесь оригинального обмена технологиями, ваша поддержка побудит меня продолжать творить! награда

WeChat награда

Вознаграждение Alipay