सीएसवी डेटा को इलास्टिक्स खोज में लोड करने के लिए लॉगस्टैश का उपयोग करना

मूल रूप से 10 सितंबर, 2015 को ObjectRocket.com/blog पर प्रकाशित किया गया

क्या आपके पास बिल्कुल नया Elasticsearch® इंस्टेंस है, लेकिन आपका सारा उपयोगी डेटा जिसे आप खोजना चाहते हैं, एक CSV में रहता है फ़ाइल?कोई बात नहीं। Logstash® लगभग किसी भी डेटा को Elasticsearch इंडेक्स में आसानी से खोजने योग्य चीज़ में बदल देता है।

आरंभ करने के लिए, आपको इन उदाहरणों का उपयोग करने के लिए कुछ डेटा और यूनिक्स® के समान वातावरण की आवश्यकता है। Windows® कुछ मामूली समायोजनों के साथ ठीक काम करता है। इस मामले में, हम अपने Davis Vantage Pro2® से डेटा का निर्यात लेना चाहते थे। वेदरस्टेशन, .CSV . में स्वरूपित करें, और इसके साथ एक नई अनुक्रमणिका बनाएं।

हमने इन जैसी कुछ मिलियन पंक्तियों के साथ शुरुआत की, जो एक स्थानीय फ़ाइल में संग्रहीत हैं:

$ head -3 /home/erik/weather.csv
HumOut,TempIn,DewPoint,HumIn,WindDir,RainMonth,WindSpeed,RainDay,BatteryVolts,WindChill,Pressure,time,TempOut,WindSpeed10Min,RainRate
76,78.0,78.227017302825,44,109,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:29Z,86.5,1,0.0
76,78.0,78.227017302825,44,107,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:45Z,86.5,1,0.0
76,78.0,78.32406784157725,44,107,2.0,0,0.0,1.236328125,90.83340000000001,29.543,2015-06-18T17:50:00Z,86.59999999999999,1,0.0

नोट: इस प्रयोग के काम करने के लिए, आपके पास कम से कम एक डेटा स्रोत होना चाहिए।

आपके पास डेटा होने के बाद, आप आरंभ कर सकते हैं। सबसे पहले, सुनिश्चित करें कि आपके पास जावा का एक संस्करण स्थापित है:

$ java -version
openjdk version "1.8.0_51"

कोई भी जावा वर्चुअल मशीन (JVM) इसके लिए ठीक है—OpenJDK®, Oracle®, इत्यादि।

$ curl -O https://download.elastic.co/logstash/logstash/logstash-1.5.4.tar.gz
$ tar xfz logstash-1.5.4.tar.gz
$ cd logstash-1.5.4
$ mkdir conf

अब, कॉन्फ़िगरेशन फ़ाइल बनाने का समय आ गया है।

सबसे पहले, एक input परिभाषित करें अनुभाग जहां आप लॉगस्टैश tell बताते हैं डेटा कहां खोजें:

input {
    file {
        path => "/home/erik/weather.csv"
        start_position => beginning

    }
}

यह केवल लॉगस्टैश tells बताता है कहाँ देखना है और हम फ़ाइल की शुरुआत से लोड करना चाहते हैं। इसके बाद, हमें एक फ़िल्टर की आवश्यकता है—लॉगस्टैश में डिफ़ॉल्ट रूप से ढेर सारे फ़िल्टर प्लग इन उपलब्ध हैं। यह उदाहरण डेटा को पार्स करने के लिए जोड़े का उपयोग करता है। अब तक, लॉगस्टैश फ़ाइल में डेटा के बारे में कुछ भी नहीं जानता है—आपको विभिन्न क्षेत्रों को संभालने के तरीके पर प्रारूप और अन्य विशिष्टताओं को निर्दिष्ट करने की आवश्यकता है:

filter {
    csv {
        columns => [
          "HumOut",
          "TempIn",
          "DewPoint",
          "HumIn",
          "WindDir",
          "RainMonth",
          "WindSpeed",
          "RainDay",
          "BatteryVolts",
          "WindChill",
          "Pressure",
          "time",
          "TempOut",
          "WindSpeed10Min",
          "RainRate"
        ]
        separator => ","
        remove_field => ["message"]
        }
    date {
        match => ["time", "ISO8601"]
    }
    mutate {
        convert => ["TempOut", "float"]
    }
}

कॉलम स्व-व्याख्यात्मक हैं, लेकिन यहां अधिक विवरण दिया गया है। सबसे पहले, उदाहरण message . को हटाता है फ़ील्ड, जो पूरी पंक्ति वाली एक प्रविष्टि है। आपको इसकी आवश्यकता नहीं होगी क्योंकि आप विशिष्ट विशेषताओं की खोज कर रहे हैं। दूसरा, यह निर्दिष्ट करता है कि time फ़ील्ड में ISO8601-formatted date . शामिल है ताकि इलास्टिक्स खोज जानता है कि यह एक सादा स्ट्रिंग नहीं है। अंत में, यह TempOut . को कन्वर्ट करने के लिए mutate फ़ंक्शन का उपयोग करता है मान को फ़्लोटिंग-पॉइंट नंबर में बदल दें।

अब, डेटा को अंतर्ग्रहण करने के लिए निम्न कोड का उपयोग करें और इसे Elasticsearch में संग्रहीत करने के बाद पार्स करें:

output {
    elasticsearch {
        protocol => "https"
        host => ["iad1-20999-0.es.objectrocket.com:20999"]
        user => "erik"
        password => "mysupersecretpassword"
        action => "index"
        index => "eriks_weather_index"
    }
    stdout { }
}

अंत में, अपने होस्ट और पोर्ट, प्रमाणीकरण डेटा और इसे स्टोर करने के लिए इंडेक्स के नाम को कॉन्फ़िगर करें।

ठीक है, चलो इसे आग लगाते हैं। अगर यह काम कर रहा है, तो इसे इस तरह दिखना चाहिए:

$ bin/logstash -f conf/logstash.conf -v
Logstash startup completed

काम किया? इलास्टिक्स खोज पूछें:

$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/_cat/indices?v'
health status index               pri rep docs.count store.size pri.store.size
green  open   eriks_weather_index 5   1   294854     95.8mb     48.5mb

दस्तावेज़ वहाँ हैं, इसलिए एक के लिए पूछें:

$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/eriks_weather_index/_search?q=TempOut:>75&pretty&terminate_after=1'

यह Elasticsearch को TempOut . के साथ दस्तावेज़ खोजने के लिए कहता है 75 से अधिक (Tempout:>75 ), इसे मानव उपभोग (सुंदर) के लिए प्रारूपित करने के लिए, और प्रति शार्प एक से अधिक परिणाम नहीं लौटाने के लिए (terminate_after=1 ) इसे कुछ इस तरह वापस करना चाहिए:

{
  "took" : 4,
  "timed_out" : false,
  "terminated_early" : true,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "failed" : 0
  },
  "hits" : {
     "total" : 5,
     "max_score" : 1.0,
       "hits" : [ {
    "_index" : "eriks_weather_index",
      "_type" : "logs",
      "_id" : "AU-yXZJIJb3HnhKvpdNC",
      "_score" : 1.0,
      "_source":{"@version":"1","@timestamp":"2015-06-22T10:24:23.000Z","host":"kibana","path":"/home/erik/weather.csv","HumOut":"86","TempIn":"79.7","DewPoint":"70.65179649787358","HumIn":"46","WindDir":"161","RainMonth":"2.7","WindSpeed":"0","RainDay":"0.36","BatteryVolts":"1.125","WindChill":"82.41464999999999","Pressure":"29.611","time":"2015-06-22T10:24:23Z","TempOut":75.1,"WindSpeed10Min":"0","RainRate":"0.0"}
    } ]
   } 
}

सफलता। लॉगस्टैश एक बेहतरीन स्विस आर्मी नाइफ है जो आपके पास रखे हुए किसी भी डेटा को ऐसी चीज़ में बदलने के लिए है जिसे आप इलास्टिक्स खोज में आसानी से चला सकते हैं, इसलिए मज़े करें!

www.rackspace.com पर जाएं और बिक्री चैट . पर क्लिक करें बातचीत शुरू करने के लिए। प्रतिक्रिया का प्रयोग करें कोई टिप्पणी करने या प्रश्न पूछने के लिए टैब।

रैकस्पेस क्लाउड सेवा की शर्तें देखने के लिए यहां क्लिक करें।