Filebeat और Elasticsearch इंजेस्ट पाइपलाइन्स के साथ CSV फाइलों को सहजता से पार्स करें

Elasticsearch 5 में सबसे अच्छी नई सुविधाओं में से एक इंजेस्ट नोड है, जो Elasticsearch क्लस्टर में कुछ Logstash-शैली प्रोसेसिंग जोड़ता है, ताकि डेटा को इंडेक्स करने से पहले किसी अन्य सर्विस और/या इंफ्रास्ट्रक्चर की आवश्यकता के बिना ट्रांसफॉर्म किया जा सके। थोड़ी देर पहले, हमने Logstash के साथ CSV फाइलों को पार्स करने पर एक त्वरित ब्लॉग पोस्ट किया था, इसलिए मैं तुलना के लिए उस का इंजेस्ट पाइपलाइन संस्करण प्रदान करना चाहूंगा।

यहां हम एक उदाहरण दिखाएंगे जिसमें Filebeat का उपयोग करके डेटा को एक इंजेस्ट पाइपलाइन में शिप किया जाता है, इसे इंडेक्स किया जाता है, और Kibana के साथ विज़ुअलाइज़ किया जाता है।

डेटा

वहां मुफ्त डेटा के लिए बहुत सारे बेहतरीन स्रोत हैं, लेकिन चूंकि ObjectRocket में हम में से अधिकांश ऑस्टिन, TX में हैं, हम data.austintexas.gov से कुछ डेटा का उपयोग करने जा रहे हैं। रेस्तरां निरीक्षण डेटा सेट एक अच्छा आकार का डेटा सेट है जिसमें हमें एक वास्तविक दुनिया का उदाहरण देने के लिए पर्याप्त प्रासंगिक जानकारी है।

नीचे इस डेटा सेट से कुछ लाइनें हैं ताकि आपको डेटा की संरचना का अंदाजा हो सके:

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... यह एक अच्छा, अनुकूल, प्रति एंट्री सिंगल लाइन केस नहीं होने जा रहा है, लेकिन यह ठीक है। जैसा कि आप देखने वाले हैं, Filebeat में मल्टीलाइन एंट्रीज को संभालने और डेटा में दबे न्यूलाइन्स के आसपास काम करने की कुछ अंतर्निहित क्षमता है।

संपादकीय नोट: मैं कुछ "अड़चनों" के साथ एक अच्छा सरल उदाहरण योजना बना रहा था, लेकिन अंत में, मुझे लगा कि इन परिदृश्यों के आसपास काम करने के लिए Elastic Stack आपको जो टूल्स देता है, उनमें से कुछ को देखना दिलचस्प हो सकता है।

Filebeat सेट अप करना

पहला कदम Filebeat को अपने Elasticsearch क्लस्टर में डेटा शिप करना शुरू करने के लिए तैयार करना है। एक बार जब आप Filebeat डाउनलोड कर लेते हैं (अपने ES क्लस्टर के समान संस्करण का उपयोग करने का प्रयास करें) और निकालते हैं, तो शामिल filebeat.yml कॉन्फ़िगरेशन फाइल के माध्यम से इसे सेट अप करना बेहद सरल है। हमारे परिदृश्य के लिए, यहां वह कॉन्फ़िगरेशन है जिसका मैं उपयोग कर रहा हूं।

filebeat.prospectors:
- input_type: log
  paths:
    - /Path/To/logs/*.csv
  # Ignore the first line with column headings
  exclude_lines: ["^Restaurant Name,"]
  # Identifies the last two columns as the end of an entry and then prepends the previous lines to it
  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"

यहां सब कुछ काफी सीधा है; आपके पास इनपुट फाइलों को कहां और कैसे हथियाना है यह निर्दिष्ट करने के लिए एक खंड है और डेटा कहां शिप करना है यह निर्दिष्ट करने के लिए एक खंड है। जिन भागों को मैं विशेष रूप से कॉल करूंगा वे मल्टीलाइन बिट और Elasticsearch कॉन्फ़िगरेशन टुकड़ा हैं।

चूंकि इस डेटा सेट के लिए फॉर्मेटिंग सुपर सख्त नहीं है, डबल कोट्स के असंगत उपयोग और कई न्यूलाइन्स छिड़के हुए हैं, सबसे अच्छा विकल्प एक एंट्री के अंत की तलाश करना था, जिसमें एक न्यूमेरिक ID होता है जिसके बाद इंस्पेक्शन प्रकार होता है बिना अधिक भिन्नता या डबल-कोट्स/न्यूलाइन्स के। वहां से, Filebeat बस किसी भी अनमैच्ड लाइन्स को कतारबद्ध करेगा और उन्हें पैटर्न से मेल खाने वाली अंतिम लाइन से पहले जोड़ देगा। यदि आपका डेटा क्लीनर है और एक साधारण प्रति एंट्री लाइन फॉर्मेट पर टिका है, तो आप मल्टीलाइन सेटिंग्स को बहुत ज्यादा नजरअंदाज कर सकते हैं।

Elasticsearch आउटपुट सेक्शन को देखते हुए, यह एक छोटे से जोड़ के साथ मानक Elasticsearch सेटिंग्स है जिस पाइपलाइन का नाम आप pipeline: निर्देश के साथ उपयोग करना चाहते हैं। यदि आप ObjectRocket सर्विस पर हैं, तो आप UI में "कनेक्ट" टैब से आउटपुट स्निपेट को सिर्फ हथिया सकते हैं, जो सभी सही होस्ट्स के साथ प्री-पॉप्युलेटेड आएगा, और बस पाइपलाइन लाइन जोड़ें और अपने उपयोगकर्ता और पासवर्ड भरें। साथ ही, सुनिश्चित करें कि यदि आपने पहले से नहीं किया है तो आपके सिस्टम का IP आपके क्लस्टर के ACLs में जोड़ा गया है।

इंजेस्ट पाइपलाइन बनाना

अब जब हमारे पास इनपुट डेटा और Filebeat तैयार है, तो हम अपनी इंजेस्ट पाइपलाइन बना और ट्वीक कर सकते हैं। पाइपलाइन को जिन मुख्य कार्यों को करने की आवश्यकता है वे हैं:

• CSV सामग्री को सही फील्ड्स में विभाजित करें
• निरीक्षण स्कोर को पूर्णांक में बदलें
• @timestamp फील्ड सेट करें
• कुछ अन्य डेटा फॉर्मेटिंग को साफ करें

यहां एक पाइपलाइन है जो यह सब कर सकती है:

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 - "
      }
    }
  ]
}

Logstash के विपरीत, इंजेस्ट पाइपलाइन में (इस लेखन के समय) CSV प्रोसेसर/प्लगइन नहीं है, इसलिए आपको CSV को खुद ट्रांसफॉर्म करना होगा। मैंने भारी उठाने के लिए एक ग्रोक प्रोसेसर का उपयोग किया, क्योंकि प्रत्येक पंक्ति में केवल कुछ कॉलम थे। कई और कॉलम वाले डेटा के लिए, ग्रोक प्रोसेसर काफी बालों वाला हो सकता है, इसलिए एक अन्य विकल्प स्प्लिट प्रोसेसर और कुछ पेनलेस स्क्रिप्टिंग का उपयोग करके लाइन को अधिक पुनरावृत्त फैशन में प्रोसेस करना है। आप दूसरे ग्रोक प्रोसेसर को भी देख सकते हैं, जो सिर्फ इस डेटा सेट में दो अलग-अलग तरीकों से दर्ज किए गए ज़िप कोड से निपटने के लिए है।

डीबग उद्देश्यों के लिए, मैंने एक सामान्य on_failure सेक्शन शामिल किया जो सभी त्रुटियों को पकड़ लेगा और प्रिंट करेगा कि किस प्रकार का प्रोसेसर विफल हुआ और वह संदेश क्या है जिसने पाइपलाइन को तोड़ा। यह डीबग को बहुत आसान बनाता है। मैं बस अपने इंडेक्स को किसी भी डॉक्यूमेंट के लिए क्वेरी कर सकता हूं जिनमें error सेट है और फिर सिमुलेट API के साथ डीबग कर सकता हूं। इस पर अभी और...

पाइपलाइन का परीक्षण

अब जब हमने अपनी इंजेस्ट पाइपलाइन कॉन्फ़िगर कर ली है, तो आइए सिमुलेट API के साथ इसका परीक्षण करें और चलाएं। सबसे पहले आपको एक नमूना डॉक्यूमेंट की आवश्यकता होगी। आप इसे कुछ तरीकों से कर सकते हैं। आप या तो पाइपलाइन सेटिंग के बिना Filebeat चला सकते हैं और फिर Elasticsearch से अनप्रोसेस्ड डॉक्यूमेंट हथिया सकते हैं, या आप कंसोल आउटपुट सक्षम करके Filebeat चला सकते हैं, Elasticsearch सेक्शन को कमेंट करके और yml फाइल में निम्नलिखित जोड़कर:

output.console:
  pretty: true

यहां मेरे वातावरण से हथियाया गया एक नमूना डॉक्यूमेंट है:

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"
      }
    }
  ]
}

और प्रतिक्रिया (मैंने इसे उन फील्ड्स तक ट्रिम कर दिया है जिन्हें हम सेट करने की कोशिश कर रहे थे):

{
  "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"
        }
      }
    }
  ]
}

पाइपलाइन निश्चित रूप से सफल रही, लेकिन सबसे महत्वपूर्ण बात, सभी डेटा सही जगह पर दिखाई देता है।

Filebeat चलाना

Filebeat चलाने से पहले, हम एक आखिरी चीज करेंगे। यह भाग पूरी तरह से वैकल्पिक है यदि आप केवल इंजेस्ट पाइपलाइन के साथ सहज होना चाहते हैं, लेकिन यदि आप उस Location फील्ड का उपयोग करना चाहते हैं जिसे हमने ग्रोक प्रोसेसर में जियो-पॉइंट के रूप में सेट किया है, तो आपको filebeat.template.json फाइल में मैपिंग जोड़नी होगी, प्रॉपर्टीज सेक्शन में निम्नलिखित जोड़कर:

"Location": {
  "type": "geo_point"
},

अब जब वह रास्ते से हट गया है, तो हम ./filebeat -e -c filebeat.yml -d "elasticsearch" चलाकर Filebeat को फायर कर सकते हैं।

डेटा का उपयोग करना

GET /filebeat-*/_count
{}
{
  "count": 25081,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  }
}

यह एक अच्छा संकेत है! आइए देखें कि क्या हमारे पास कोई त्रुटि थी:

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": []
  }
}

एक और अच्छा संकेत!

अब हम Kibana में अपने डेटा को विज़ुअलाइज़ और दिखाने के लिए पूरी तरह तैयार हैं। हम दूसरी बार Kibana डैशबोर्ड बनाने के माध्यम से चल सकते हैं, लेकिन चूंकि हमारे पास तारीखें, रेस्तरां नाम, स्कोर और स्थान हैं, हमारे पास कुछ कूल विज़ुअलाइज़ेशन बनाने के लिए बहुत स्वतंत्रता है।

अंतिम नोट्स

एक बार फिर, इंजेस्ट पाइपलाइन काफी शक्तिशाली है और ट्रांसफॉर्मेशन को काफी आसानी से संभाल सकती है। आप अपने सभी प्रोसेसिंग को Elasticsearch में स्थानांतरित कर सकते हैं और अपने होस्ट्स पर केवल लाइटवेट Beats का उपयोग कर सकते हैं, बिना पाइपलाइन में कहीं Logstash की आवश्यकता के। हालांकि, इंजेस्ट नोड की तुलना में Logstash में अभी भी कुछ अंतराल हैं। उदाहरण के लिए, इंजेस्ट पाइपलाइन में उपलब्ध प्रोसेसर की संख्या अभी भी सीमित है, इसलिए CSV को पार्स करने जैसे सरल कार्य Logstash में जितने आसान नहीं हैं। Elasticsearch टीम नियमित रूप से नए प्रोसेसर रोल आउट करती दिख रही है, इसलिए यहां उम्मीद है कि अंतरों की सूची छोटी होती रहेगी।