Coolest New Feature ใน Elasticsearch 5 = Ingest Node เพิ่ม Logstash-style Processing ให้ ES Cluster Transform Data ก่อน Index โดยไม่ต้อง Service/Infrastructure อื่น บทความก่อนหน้า Parse CSV ด้วย Logstash คราวนี้ Ingest Pipeline Version สำหรับ Compare
แสดง Example: Filebeat Ship Data -> Ingest Pipeline -> Index -> Visualize Kibana
The Data
Free Data Sources เยอะ แต่อยู่ Austin, TX -> ใช้ Data จาก data.austintexas.gov Restaurant Inspection Dataset Good Size + Relevant Info Real-world Example
Sample Lines:
Restaurant Name,Zip Code,Inspection Date,Score,Address,Facility ID,Process Description Westminster Manor,78731,07/21/2015,96,"4100 JACKSON AVE AUSTIN, TX 78731 (30.314499, -97.755166)",2800365,Routine Inspection Wieland Elementary,78660,10/02/2014,100,"900 TUDOR HOUSE RD AUSTIN, TX 78660 (30.422862, -97.640183)",10051637,Routine Inspection DOH... ไม่ใช่ Friendly Single Line Per Entry แต่อยู่ๆ Filebeat มี Built-in Multiline Handling Work Around Buried Newlines
Editorial Note: Plan Nice Simple Example แต่สุดท้ายคิดว่า Interesting ที่จะ見 Tools ที่ Elastic Stack ให้ Work Around Scenarios นี้
Setting up Filebeat
Step 1: Filebeat Ready Ship Data ไป ES Cluster Download Filebeat (Same Version As ES Cluster) Extract Setup Simple Via filebeat.yml Config สำหรับ Scenario นี้ Config ที่ใช้:
filebeat.prospectors: - input_type: log paths: - /Path/To/logs/*.csv # Ignore first line with column headings exclude_lines: ["^Restaurant Name,"] # Identifies last two columns as end of entry prepend previous lines multiline.pattern: ',\d+,[^\",]+$' multiline.negate: true multiline.match: before #================================ Outputs ===================================== output.elasticsearch: # Array of hosts to connect to. hosts: ["https://dfw-xxxxx-0.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-1.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-2.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-3.es.objectrocket.com:xxxxx"] pipeline: "inspectioncsvs" # Optional protocol and basic auth credentials. username: "esuser" password: "supersecretpassword" Straightforward: Input Section + Output Section 2 ส่วนเรียก Multiline และ Elasticsearch Config
Multiline: Data Set Format ไม่ Strict Inconsistent Double Quotes + Newlines Sprinkled Best Option: หา End of Entry = Numeric ID + Inspection Type ไม่ Variation/Quotes/Newlines Filebeat Queue Unmatched Lines Prepend ไป Final Line Match Pattern Data Clean Single Line/Entry -> Ignore Multiline Settings
Elasticsearch Output: Standard Settings + Pipeline Directive Name หาก ObjectRocket -> Grab Output Snippet จาก "Connect" Tab UI Prepopulated Hosts Add Pipeline Line Fill User/Pass เช็ค System IP ใน Cluster ACLs
Creating the Ingest Pipeline
Input Data + Filebeat Ready -> Create/Tweak Ingest Pipeline Main Tasks:
- Split CSV Content -> Correct Fields
- Convert Inspection Score -> Integer
- Set @timestamp Field
- Clean Up Data Formatting
Pipeline ที่ทำได้ทั้งหมด:
PUT _ingest/pipeline/inspectioncsvs { "description" : "Convert Restaurant inspections csv data to indexed data", "processors" : [ { "grok": { "field": "message", "patterns": ["%{REST_NAME:RestaurantName},%{REST_ZIP:ZipCode},%{MONTHNUM2:InspectionMonth}/%{MONTHDAY:InspectionDay}/%{YEAR:InspectionYear},%{NUMBER:Score},\"%{DATA:StreetAddress}\n%{DATA:City},?\\s+%{WORD:State}\\s*%{NUMBER:ZipCode2}\\s*\n\\(?%{DATA:Location}\\)?\",%{NUMBER:FacilityID},%{DATA:InspectionType}$"], "pattern_definitions": { "REST_NAME": "%{DATA}|%{QUOTEDSTRING}", "REST_ZIP": "%{QUOTEDSTRING}|%{NUMBER}" } } }, { "grok": { "field": "ZipCode", "patterns": [".*%{ZIP:ZipCode}\"?$"], "pattern_definitions": { "ZIP": "\\d{5}" } } }, { "convert": { "field" : "Score", "type": "integer" } }, { "set": { "field" : "@timestamp", "value" : "//" } }, { "date" : { "field" : "@timestamp", "formats" : ["yyyy/MM/dd"] } } ], "on_failure" : [ { "set" : { "field" : "error", "value" : " - Error processing message - " } } ] } Ingest Pipeline ไม่มี CSV Processor/Plugin (Time of Writing) Transform CSV เอง ใช้ Grok Processor Heavy Lifting (Few Columns) Many Columns -> Grok Hairy Option: Split Processor + Painless Scripting Process Line Iteratively Second Grok: Deal With 2 Ways Zip Codes Entered
Debug: on_failure Section Catch All Errors Print Failed Processor Type + Broken Message Debug Easier Query Index Error Set -> Debug Simulate API
Testing the Pipeline
Pipeline Configured -> Test/Run Simulate API Sample Document: Run Filebeat Without Pipeline Setting Grab Unprocessed Doc จาก ES หรือ Run Filebeat Console Output (Comment ES Section Add output.console pretty: true)
Sample Document:
POST _ingest/pipeline/inspectioncsvs/_simulate { "docs" : [ { "_index": "inspections", "_type": "log", "_id": "AVpsUYR_du9kwoEnKsSA", "_score": 1, "_source": { "@timestamp": "2017-03-31T18:22:25.981Z", "beat": { "hostname": "systemx", "name": "RestReviews", "version": "5.1.1" }, "input_type": "log", "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection", "offset": 2109798, "source": "/Path/to/my/logs/Restaurant_Inspection_Scores.csv", "tags": [ "debug", "reviews" ], "type": "log" } } ] } Response (Trimmed Fields We Set):
{ "docs": [ { "doc": { "_id": "AVpsUYR_du9kwoEnKsSA", "_type": "log", "_index": "inspections", "_source": { "InspectionType": "Routine Inspection", "ZipCode": "78660", "InspectionMonth": "10", "City": "AUSTIN", "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection", "RestaurantName": "Wieland Elementary", "FacilityID": "10051637", "Score": 100, "StreetAddress": "900 TUDOR HOUSE RD", "State": "TX", "InspectionDay": "02", "InspectionYear": "2014", "ZipCode2": "78660", "Location": "30.422862, -97.640183" }, "_ingest": { "timestamp": "2017-03-31T20:36:59.574+0000" } } } ] } Pipeline Succeeded Data In Right Place
Running filebeat
Before Run Filebeat: 1 Last Thing Optional หากใช้ Location Field (Grok Processor) As Geo-point -> Add Mapping filebeat.template.json Properties Section:
"Location": { "type": "geo_point" }, Run ./filebeat -e -c filebeat.yml -d "elasticsearch"
Using the Data
GET /filebeat-*/_count {} { "count": 25081, "_shards": { "total": 5, "successful": 5, "failed": 0 } } Good Sign! Check Errors:
GET /filebeat-*/_search { "query": { "exists" : { "field" : "error" } } } { "took": 1, "timed_out": false, "_shards": { "total": 5, "successful": 5, "failed": 0 }, "hits": { "total": 0, "max_score": null, "hits": [] } } Another Good Sign! Ready Visualize Kibana Walkthrough Dashboards Another Time แต่มี Dates, Restaurant Names, Scores, Locations -> Freedom Create Cool Visualizations
Final Notes
Ingest Pipeline Powerful Handle Transformations Easy Move All Processing To ES Lightweight Beats On Hosts No Logstash Required แต่ยัง Gaps vs Logstash: Processors Limited Simple CSV Parsing ไม่ Easy Like Logstash ES Team Roll Out New Processors Regularly Hope Differences Keep Shrinking
