นำเข้าข้อมูล CSV เข้า Elasticsearch ได้อย่างง่ายดายด้วย Logstash

เผยแพร่ครั้งแรกเมื่อวันที่ 10 กันยายน 2015 ที่ ObjectRocket.com/blog

คุณมีอินสแตนซ์ Elasticsearch® ใหม่ แต่ข้อมูลที่มีประโยชน์ทั้งหมดที่คุณต้องการค้นหาอยู่ในไฟล์ CSV? ไม่มีปัญหา Logstash® ทำให้ข้อมูลเกือบทุกประเภทกลายเป็นข้อมูลที่ค้นหาได้ง่ายในดัชนี Elasticsearch

เพื่อเริ่มต้น คุณต้องมีข้อมูลและสภาพแวดล้อมคล้าย Unix® เพื่อใช้ตัวอย่างเหล่านี้ 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 แล้ว:

$ java -version
openjdk version "1.8.0_51"

Java Virtual Machine (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

ตอนนี้ถึงเวลาสร้างไฟล์การตั้งค่า (configuration file)

ขั้นแรก กำหนดส่วน input เพื่อบอก Logstash ว่าจะหาข้อมูลที่ไหน:

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

นี่แค่บอก Logstash ว่าจะดูที่ไหน และเราต้องการโหลดตั้งแต่ต้นไฟล์ ต่อไป เราต้องการตัวกรอง (filter)—Logstash มีปลั๊กอินตัวกรองมากมายให้ใช้งานได้ทันที ตัวอย่างนี้ใช้สองตัวเพื่อแยกวิเคราะห์ข้อมูล จนถึงตอนนี้ Logstash ยังไม่รู้อะไรเกี่ยวกับข้อมูลในไฟล์—คุณต้องระบุรูปแบบและรายละเอียดอื่นๆ ว่าจะจัดการฟิลด์ต่างๆ อย่างไร:

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 เพื่อให้ Elasticsearch รู้ว่าไม่ใช่สตริงธรรมดา ข้อสาม ใช้ฟังก์ชัน mutate เพื่อแปลงค่า TempOut ให้เป็นตัวเลขทศนิยม (floating-point)

ตอนนี้ ใช้โค้ดต่อไปนี้เพื่อนำเข้า (ingest) ข้อมูลและแยกวิเคราะห์หลังจากจัดเก็บใน 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

ทำงานได้ไหม? ถาม Elasticsearch:

$ 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

เอกสารอยู่ที่นั่นแล้ว ลองคิวรี (query) หนึ่งดู:

$ 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) จัดรูปแบบให้มนุษย์อ่านง่าย (pretty) และคืนค่าผลลัพธ์ไม่เกินหนึ่งต่อชาร์ด (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"
        }
      }
    ]
  }
}

สำเร็จ! Logstash เป็น Swiss Army Knife ที่ยอดเยี่ยมสำหรับการเปลี่ยนข้อมูลที่คุณมีอยู่ให้กลายเป็นสิ่งที่คุณเล่นกับ Elasticsearch ได้ง่าย ดังนั้นสนุกกับมันไปเลย!

เยี่ยมชม www.rackspace.com และคลิก Sales Chat เพื่อเริ่มการสนทนา ใช้แท็บ Feedback เพื่อแสดงความคิดเห็นหรือถามคำถาม

คลิกที่นี่เพื่อดู ข้อกำหนดบริการ Rackspace Cloud