Breizhcamp 2017 : Ingest node : (ré)indexer et enrichir des documents dans Elasticsearch

Transcript

1. ‹#› Ingest Node is powered by (re)indexing and enriching documents

   within Elasticsearch David Pilato Developer | Evangelist @dadoonet

2. ‹#›

3. 3 infom ercial

4. 4 The only Elasticsearch as a Service offering powered by

   the creators of the Elastic Stack • Always runs on the latest software • One-click to scale/upgrade with no downtime • Free Kibana and backups every 30 minutes • Dedicated, SLA-based support • Easily add X-Pack features: security (Shield), alerting (Watcher), and monitoring (Marvel) • Pricing starts at $45 a month infom ercial

5. sli.do/elastic

6. ‹#› Why ingest node?

7. ‹#› I just want to tail a file.

8. sli.do/elastic Logstash: collect, enrich & transport 8 grok date mutate

   input output Filters The file Elasticsearch

9. sli.do/elastic Logstash common setup 9 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200]

   "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 message

10. sli.do/elastic Or … 10 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET

    /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 message

11. sli.do/elastic Ingest node setup 11 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200]

    "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:01 +0200] "GET /cgi-bin/try/ HTTP/1.1" 200 3395 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:18 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:00 +0200] "GET /robots.txt HTTP/1.1" 200 68 127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 127.0.0.1 - - [19/Apr/2016:12:00:15 +0200] "GET / HTTP/1.1" 200 24

12. sli.do/elastic Filebeat: collect and ship 12 127.0.0.1 - - [19/Apr/2016:12:00:04

    +0200] "GET / HTTP/1.1" 200 24 127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] "GET /not_found/ HTTP/1.1" 404 7218 127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] "GET /favicon.ico HTTP/1.1" 200 3638 { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:04 +0200] \"GET / HTTP/1.1\" 200 24" } { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:07 +0200] \"GET /not_found/ HTTP/1.1\" 404 7218" } { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:09 +2000] \"GET /favicon.ico HTTP/1.1\" 200 3638" }

13. sli.do/elastic Elasticsearch: enrich and index 13 { "message" : "127.0.0.1

    - - [19/Apr/2016:12:00:04 +0200] \"GET / HTTP/1.1\" 200 24" } { "request" : "/", "auth" : "-", "ident" : "-", "verb" : "GET", "@timestamp" : "2016-04-19T10:00:04.000Z", "response" : "200", "bytes" : "24", "clientip" : "127.0.0.1", "httpversion" : "1.1", "rawrequest" : null, "timestamp" : "19/Apr/2016:12:00:04 +0200" }

14. ‹#› How does ingest node work?

15. sli.do/elastic Ingest pipeline 15 Pipeline: a set of processors grok

    date remove document enriched document

16. sli.do/elastic 16 grok remove attachment convert uppercase foreach trim append

    gsub set split fail geoip join lowercase rename date

17. Extracts structured fields out of a single text field 17

    Grok processor { "grok": { "field": "message", "patterns": ["%{DATE:date}"] } }

18. set, remove, rename, convert, gsub, split, join, lowercase, uppercase, trim,

    append 18 Mutate processors { "remove": { "field": "message" } }

19. Parses a date from a string 19 Date processor {

    "date": { "field": "timestamp", "formats": ["YYYY"] } }

20. Adds information about the geographical location of IP addresses 20

    Geoip processor { "geoip": { "field": "ip" } }

21. Introducing new processors is as easy as writing a plugin

    21 Plugins { "your_plugin": { ... } }

22. sli.do/elastic Define a pipeline PUT /_ingest/pipeline/apache-log { "processors" : [

    { "grok" : { "field": "message", "patterns": ["%{COMMONAPACHELOG}"] } }, { "date" : { "field" : "timestamp", "formats" : ["dd/MMM/YYYY:HH:mm:ss Z"] } }, { "remove" : { "field" : "message" } } ] } 22

23. sli.do/elastic PUT /_ingest/pipeline/apache-log { ... } GET /_ingest/pipeline/apache-log GET /_ingest/pipeline/*

    DELETE /_ingest/pipeline/apache-log Pipeline management Create, Read, Update & Delete 23

24. ‹#› Where can ingest pipelines be used?

25. 25 Index api PUT /logs/apache/1?pipeline=apache-log { "message" : "..." }

26. 26 Bulk api PUT /logs/_bulk { "index": { "_type": "apache",

    "_id": "1", "pipeline": "apache-log" } }\n { "message" : "..." }\n { "index": {"_type": "mysql", "_id": "1", "pipeline": "mysql-log" } }\n { "message" : "..." }\n

27. Scroll & bulk indexing made easy 27 Reindex api POST

    /_reindex { "source": { "index": "logs", "type": "apache" }, "dest": { "index": "apache-logs", "pipeline" : "apache-log" } }

28. ‹#› Error handling

29. sli.do/elastic 29 grok date remove { "message" : "127.0.0.1 -

    - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

30. sli.do/elastic 30 grok date remove 400 Bad Request unable to

    parse date [19/Apr/2016:12:00:00 +040] { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

31. sli.do/elastic 31 grok date remove set on failure processors at

    the pipeline level { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

32. sli.do/elastic 32 remove 200 OK grok date set on failure

    processors at the pipeline level { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

33. sli.do/elastic 33 grok date remove set on failure processors at

    the processor level remove { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

34. sli.do/elastic 34 grok date remove set remove 200 OK on

    failure processors at the processor level { "message" : "127.0.0.1 - - [19/Apr/2016:12:00:00 +040] \"GET / HTTP/1.1\" 200 24" }

35. ‹#› Ingest node internals

36. sli.do/elastic cluster Default scenario 36 Client node1 logs 2P logs

    3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS Cluster State logs index: 3 primary shards, 1 replica each All nodes are equal: - node.data: true - node.master: true - node.ingest: true

37. sli.do/elastic cluster Default scenario 37 Client node1 logs 2P logs

    3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS Pre-processing on the coordinating node All nodes are equal: - node.data: true - node.master: true - node.ingest: true index request for shard 3

38. sli.do/elastic cluster Default scenario 38 Client node1 logs 2P logs

    3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS Indexing on the primary shard All nodes are equal: - node.data: true - node.master: true - node.ingest: true index request for shard 3

39. sli.do/elastic cluster Default scenario 39 Client node1 logs 2P logs

    3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS Indexing on the replica shard All nodes are equal: - node.data: true - node.master: true - node.ingest: true index request for shard 3

40. sli.do/elastic cluster Ingest dedicated nodes 40 Client node1 logs 2P

    logs 3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS node4 CS node5 CS node.data: false node.master: false node.ingest: true node.data: true node.master: true node.ingest: false

41. sli.do/elastic cluster Ingest dedicated nodes 41 Client node1 logs 2P

    logs 3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS node4 CS node5 CS index request for shard 3 Forward request to an ingest node

42. sli.do/elastic cluster Ingest dedicated nodes 42 Client node1 logs 2P

    logs 3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS node4 CS node5 CS index request for shard 3 Pre-processing on the ingest node

43. sli.do/elastic cluster Ingest dedicated nodes 43 Client node1 logs 2P

    logs 3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS node4 CS node5 CS index request for shard 3 Indexing on the primary shard

44. sli.do/elastic cluster Ingest dedicated nodes 44 Client node1 logs 2P

    logs 3R CS node2 logs 3P logs 1R CS node3 logs 1P logs 2R CS node4 CS node5 CS index request for shard 3 Indexing on the replica shard

45. ‹#› From postal address to geo_point From geo_point to postal

    address bano-ingest plugin

46. sli.do/elastic What is BANO? • French Open Data base for

    postal addresses • http://openstreetmap.fr/bano • http://bano.openstreetmap.fr/data/ 46 per department all addresses

47. sli.do/elastic BANO Format 47 • bano-976.csv sample (full.csv.gz has same

    format) 976030950H-26,26,RUE DISMA,97660,Bandrélé,CAD,-12.891701,45.202652
 976030950H-28,28,RUE DISMA,97660,Bandrélé,CAD,-12.891900,45.202700
 976030950H-30,30,RUE DISMA,97660,Bandrélé,CAD,-12.891781,45.202535
 976030950H-32,32,RUE DISMA,97660,Bandrélé,CAD,-12.892005,45.202564
 976030950H-3,3,RUE DISMA,97660,Bandrélé,CAD,-12.892444,45.202135
 976030950H-34,34,RUE DISMA,97660,Bandrélé,CAD,-12.892068,45.202450
 976030950H-4,4,RUE DISMA,97660,Bandrélé,CAD,-12.892446,45.202367
 976030950H-5,5,RUE DISMA,97660,Bandrélé,CAD,-12.892461,45.202248
 976030950H-6,6,RUE DISMA,97660,Bandrélé,CAD,-12.892383,45.202456
 976030950H-8,8,RUE DISMA,97660,Bandrélé,CAD,-12.892300,45.202555
 976030950H-9,9,RUE DISMA,97660,Bandrélé,CAD,-12.892355,45.202387 976030951J-103,103,RTE NATIONALE 3,97660,Bandrélé,CAD,-12.893639,45.201696 \_ ID | \_ Street Name | \ \_ Source \_ Geo point | | \ |_ Street Number |_ Zipcode \_ City Name

48. sli.do/elastic Features • Download, transform and index BANO datasource •

    Create a new ingest processor 48 curl -XPUT 127.0.0.1:9200/_bano/17 curl -XPUT 127.0.0.1:9200/_bano/17,95,29 curl -XPUT 127.0.0.1:9200/_bano/_full curl -XPUT "localhost:9200/_ingest/pipeline/bano-test?pretty" -d '{ "description": "my_pipeline", "processors": [ { "bano": {} } ] }'

49. sli.do/elastic From structured address (french format)… 49 curl -XPOST "localhost:9200/_ingest/pipeline/bano-test/_simulate?pretty"

    -d '{ "docs": [ { "_index": "index", "_type": "type", "_id": "id", "_source": { "address": { "number": "25", "street_name": "georges", "zipcode": "17440", "city": "Aytré" } } } ] }'

50. sli.do/elastic To normalized address with coordinates… 50 "doc" : {


    "_source" : {
 "address" : { "zipcode" : "17440",
 "number" : "25",
 "city" : "Aytré",
 "street_name" : "georges" },
 "bano_address" : {
 "zipcode" : "17440",
 "number" : "25",
 "city" : "Aytré", "street_name" : "Boulevard Georges Clemenceau",
 "full_address" : "25, Boulevard Georges Clemenceau 17440 Aytré",
 "location" : {
 "lon" : -1.122966,
 "lat" : 46.130368
 }
 }
 }
 }

51. sli.do/elastic From a geo point… 51 curl -XPOST "localhost:9200/_ingest/pipeline/bano-test/_simulate?pretty" -d

    '{ "docs": [ { "_index": "index", "_type": "type", "_id": "id", "_source": { "location": { "lat": 46.135283, "lon": -1.113750 } } } ] }'

52. sli.do/elastic To the closest full address… 52 "doc" : {


    "_source" : {
 "location" : {
 "lon" : -1.11375,
 "lat" : 46.135283
 },
 "bano_address" : {
 "zipcode" : "17440",
 "number" : "1",
 "city" : "Aytré",
 "street_name" : "Rue du Petit Versailles",
 "full_address" : "1, Rue du Petit Versailles 17440 Aytré",
 "location" : {
 "lon" : -1.113564,
 "lat" : 46.135343
 }
 }
 }
 }
 }

53. sli.do/elastic Combine with other ingest processors 53 curl -XPUT "localhost:9200/_ingest/pipeline/bano-test-4?pretty"

    -d '{
 "description": "debug",
 "processors": [ {
 "geoip" : {
 "field" : "ip"
 }
 }, {
 "bano": {
 "location_lat_field": "geoip.location.lat",
 "location_lon_field": "geoip.location.lon"
 }
 } ]
 }'

54. sli.do/elastic From an IP address… 54 curl -XPOST "localhost:9200/_ingest/pipeline/bano-test-4/_simulate?pretty&verbose" -d

    '{
 "docs": [ {
 "_index": "index",
 "_type": "type",
 "_id": "id",
 "_source": {
 "ip" : "82.229.80.187"
 }
 } ]
 }'

55. sli.do/elastic To the closest full address… 55 "doc" : {


    "_source" : {
 "ip" : "82.229.80.187",
 "geoip" : {
 "continent_name" : "Europe", "city_name" : "Cergy",
 "country_iso_code" : "FR", "region_name" : "Val d'Oise",
 "location" : { "lon" : 2.0761, "lat" : 49.0364 }
 },
 "bano_address" : {
 "zipcode" : "95000",
 "number" : "3",
 "city" : "Cergy",
 "location" : {
 "lon" : 2.075687,
 "lat" : 49.037202
 },
 "full_address" : "3, Avenue des Trois Fontaines 95000 Cergy",
 "street_name" : "Avenue des Trois Fontaines"
 }
 }
 }

56. ‹#› Demo time!

57. ‹#› Writing an ingest plugin

58. sli.do/elastic Writing the processor public final class BanoProcessor extends AbstractProcessor

    { private final String cityField; public BanoProcessor(String cityField) { this.cityField = cityField; } @Override
 public void execute(IngestDocument ingestDocument) {
 // Implement your logic code here if (ingestDocument.hasField(cityField)) { String city = ingestDocument.getFieldValue(cityField, String.class) // Like searching in elasticsearch with a city field Location location = banoEsClient.searchByCity(city); // Then modify the document as you wish Map<String, Object> locationObject = new HashMap<>();
 locationObject.put("lat", location.getLat());
 locationObject.put("lon", location.getLon());
 ingestDocument.setFieldValue("location", locationObject); } } } 58

59. sli.do/elastic Writing the processor factory 59 public static final class

    Factory implements Processor.Factory { @Override
 public Processor create(Map<String, Processor.Factory> map, String processorTag, Map<String, Object> config) throws Exception { // Read the bano processor config
 String cityField = readStringProperty("bano", processorTag, config, "city_field", // We read here the value of "city_field" in config "address.city"); // If not set we will read from "address.city" by default // Do the same for other fields // Create the processor instance return new BanoProcessor(cityField, otherFields...); }


60. sli.do/elastic Writing an ingest plugin 60 public class IngestBanoPlugin extends

    Plugin implements IngestPlugin { 
 @Override
 public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
 return Collections.singletonMap("bano", new BanoProcessor.Factory());
 } }

61. ‹#› https://www.elastic.co/downloads/elasticsearch Go get Elasticsearch 5.3.0!

62. None

63. ‹#› Watch this space: https://github.com/dadoonet And follow me on Twitter!

    Bano ingest plugin David Pilato Developer | Evangelist @dadoonet